Allocate isolated ports for service regression fixtures

Why: Concurrent task worktrees reuse fixed ports and can enter an unrelated port-occupied path instead of the intended service assertion.

Evidence: An owned listener on port 18803 reproduced the existing failure; focused, parallel, and deployment fixture checks passed with the original assertions retained.

Impact: Test fixtures only; use per-instance dynamic ports and preserve restart and stolen-port checks. Full combination validation follows.

Refs: stage 6 local verification; /tmp/rdm-svc-port-isolation-report-20261014.md

tests/test_svc.py

@@ -8,6 +8,7 @@
 """
 import os
 import shutil
+import socket
 import subprocess
 import tempfile
 import unittest
@@ -67,7 +68,7 @@ case "${1:-}" in
     # GET/HEAD 一律 200（健康探测语义与 http.server 等价）；PUT /api/* 捕获头与
     # 请求体到 trace/put.txt 供播种断言（settings 播种验证，2026-10-11）。
     # HTTP 子进程轮询 ppid：父进程被 SIGKILL 后自行退出，避免与 recover 套件
-    # 共用 1880x 端口时泄漏占用（与 test_svc_recover 桩同规）。
+    # 并行运行时泄漏端口占用（与 test_svc_recover 桩同规）。
     python3 - "$port" "$host" "$trace_dir" <<'PYSRV' >/dev/null 2>&1 &
 import os, sys, threading, time
 from http.server import BaseHTTPRequestHandler, HTTPServer
@@ -133,6 +134,7 @@ class SvcDbRecoveryTests(unittest.TestCase):
         tmp = tempfile.mkdtemp(prefix="svc-db-regress-")
         self.addCleanup(shutil.rmtree, tmp, True)
         self.root = Path(tmp)
+        self._allocated_ports = set()
         (self.root / "scripts").mkdir(parents=True)
         shutil.copy2(SVC_SCRIPT, self.root / "scripts" / "svc.sh")
         self.trace = self.root / "trace"
@@ -190,6 +192,21 @@ class SvcDbRecoveryTests(unittest.TestCase):
             self.assertEqual(proc.returncode, 0, proc.stdout)
         return proc
 
+    def available_port(self):
+        """Allocate a distinct service port for this test, including address changes.
+
+        The socket closes before svc.sh binds it, leaving a small allocation-to-start
+        race with unrelated processes. Keep allocated numbers distinct within the
+        test so stopped/restarted services never share another instance's port.
+        """
+        while True:
+            with socket.socket() as sock:
+                sock.bind(("0.0.0.0", 0))
+                port = sock.getsockname()[1]
+            if port not in self._allocated_ports:
+                self._allocated_ports.add(port)
+                return port
+
     def runtime(self, name):
         return self.root / "runtime" / name
 
@@ -259,11 +276,12 @@ class SvcDbRecoveryTests(unittest.TestCase):
     # ---- 1. 存量 exec= DSN：连续启动两次均恢复，进程参数无秘密 ----
 
     def test_legacy_exec_dsn_survives_consecutive_restarts(self):
+        port = self.available_port()
         self.register_pg("pg1", 18890, password="legacy-pw-1")
         dsn = "postgresql://agent:legacy-pw-1@127.0.0.1:18890/agent"
         d = self.runtime("dev")
-        self.seed_agent("dev", 18791,
+        self.seed_agent("dev", port,
-                        "--addr 127.0.0.1:18791 --data %s --plugins %s --auth auto --dsn %s"
+                        f"--addr 127.0.0.1:{port} --data %s --plugins %s --auth auto --dsn %s"
                         % (d / "data", d / "plugins", dsn))
         for _ in range(2):
             self.clear_serve_record()
@@ -285,11 +303,12 @@ class SvcDbRecoveryTests(unittest.TestCase):
     # ---- 2. 仅 dbref：裸 restart 与重新部署均保留绑定 ----
 
     def test_dbref_binding_survives_bare_restart_and_redeploy(self):
+        port = self.available_port()
         self.register_pg("pg1", 18891, password="bind-pw-2")
         dsn = "postgresql://agent:bind-pw-2@127.0.0.1:18891/agent"
         d = self.runtime("dev")
-        self.seed_agent("dev", 18792,
+        self.seed_agent("dev", port,
-                        "--addr 127.0.0.1:18792 --data %s --plugins %s --auth auto"
+                        f"--addr 127.0.0.1:{port} --data %s --plugins %s --auth auto"
                         % (d / "data", d / "plugins"),
                         extra_rows=[("dbref", "pg1/agent"), ("db", "pg")])
         self.clear_serve_record()
@@ -309,29 +328,32 @@ class SvcDbRecoveryTests(unittest.TestCase):
     # ---- 3. 改地址不改数据库与认证 ----
 
     def test_addr_change_keeps_database_and_auth(self):
+        old_port = self.available_port()
+        new_port = self.available_port()
         self.register_pg("pg1", 18892, password="addr-pw-3")
         dsn = "postgresql://agent:addr-pw-3@127.0.0.1:18892/agent"
         d = self.runtime("dev")
-        self.seed_agent("dev", 18793,
+        self.seed_agent("dev", old_port,
-                        "--addr 127.0.0.1:18793 --data %s --plugins %s --auth auto"
+                        f"--addr 127.0.0.1:{old_port} --data %s --plugins %s --auth auto"
                         % (d / "data", d / "plugins"),
                         extra_rows=[("dbref", "pg1/agent"), ("db", "pg")])
         self.clear_serve_record()
-        self.svc("dev", "restart", "--addr", "127.0.0.1:18794")
+        self.svc("dev", "restart", "--addr", f"127.0.0.1:{new_port}")
         record = self.serve_record()
         self.assertIn("AGENT_SYSTEM_DSN=%s" % dsn, record)
         self.assertIn("--auth auto", record)
         self.assertEqual(self.kv("dev", "dbref"), "pg1/agent")
-        self.assertEqual(self.kv("dev", "addr"), "127.0.0.1:18794")
+        self.assertEqual(self.kv("dev", "addr"), f"127.0.0.1:{new_port}")
-        self.assertIn("--addr 127.0.0.1:18794", self.kv("dev", "exec"))
+        self.assertIn(f"--addr 127.0.0.1:{new_port}", self.kv("dev", "exec"))
         self.assertEqual(self.kv("dev", "db"), "pg")
 
     # ---- 4. 外部直连与配置文件来源：后续无参数启动仍恢复 ----
 
     def test_external_dsn_and_conffile_recover_on_bare_start(self):
+        port = self.available_port()
         d = self.runtime("dev")
-        self.seed_agent("dev", 18795,
+        self.seed_agent("dev", port,
-                        "--addr 127.0.0.1:18795 --data %s --plugins %s --auth auto"
+                        f"--addr 127.0.0.1:{port} --data %s --plugins %s --auth auto"
                         % (d / "data", d / "plugins"))
         external = "postgresql://extu:ext-pw-4@10.9.8.7:5432/extdb"
         self.clear_serve_record()
@@ -369,9 +391,13 @@ class SvcDbRecoveryTests(unittest.TestCase):
     # ---- 5. 拒启场景：不降级 SQLite、不启动进程、登记可恢复、停机前发现 ----
 
     def test_refusals_keep_registration_without_sqlite_fallback(self):
+        a1_port = self.available_port()
+        a2_port = self.available_port()
+        a3_port = self.available_port()
+        a4_port = self.available_port()
         d = self.runtime("a1")
-        self.seed_agent("a1", 18796,
+        self.seed_agent("a1", a1_port,
-                        "--addr 127.0.0.1:18796 --data %s --plugins %s --auth auto"
+                        f"--addr 127.0.0.1:{a1_port} --data %s --plugins %s --auth auto"
                         % (d / "data", d / "plugins"),
                         extra_rows=[("dbref", "pgmissing/agent"), ("db", "pg")])
         self.clear_serve_record()
@@ -384,8 +410,8 @@ class SvcDbRecoveryTests(unittest.TestCase):
         # 实例登记不全（缺 password）→ 拒启。
         self.register_pg("pginc", 18895, complete=False)
         d = self.runtime("a2")
-        self.seed_agent("a2", 18797,
+        self.seed_agent("a2", a2_port,
-                        "--addr 127.0.0.1:18797 --data %s --plugins %s --auth auto"
+                        f"--addr 127.0.0.1:{a2_port} --data %s --plugins %s --auth auto"
                         % (d / "data", d / "plugins"),
                         extra_rows=[("dbref", "pginc/agent"), ("db", "pg")])
         result = self.svc("a2", "restart", check=False)
@@ -395,8 +421,8 @@ class SvcDbRecoveryTests(unittest.TestCase):
         # dbref 与 dsn= 指向不同库 → 配置冲突拒启。
         self.register_pg("pg1", 18896, password="clash-pw")
         d = self.runtime("a3")
-        self.seed_agent("a3", 18798,
+        self.seed_agent("a3", a3_port,
-                        "--addr 127.0.0.1:18798 --data %s --plugins %s --auth auto"
+                        f"--addr 127.0.0.1:{a3_port} --data %s --plugins %s --auth auto"
                         % (d / "data", d / "plugins"),
                         extra_rows=[("dbref", "pg1/agent"), ("db", "pg"),
                                     ("dsn", "postgresql://u:p@127.0.0.1:18999/otherdb")])
@@ -406,8 +432,8 @@ class SvcDbRecoveryTests(unittest.TestCase):
 
         # db=pg 但无可恢复 DSN → 拒启。
         d = self.runtime("a4")
-        self.seed_agent("a4", 18799,
+        self.seed_agent("a4", a4_port,
-                        "--addr 127.0.0.1:18799 --data %s --plugins %s --auth auto"
+                        f"--addr 127.0.0.1:{a4_port} --data %s --plugins %s --auth auto"
                         % (d / "data", d / "plugins"),
                         extra_rows=[("db", "pg")])
         result = self.svc("a4", "restart", check=False)
@@ -416,10 +442,11 @@ class SvcDbRecoveryTests(unittest.TestCase):
         self.assertEqual(self.kv("a4", "db"), "pg")
 
     def test_restart_refusal_happens_before_stopping_running_service(self):
+        port = self.available_port()
         self.register_pg("pg1", 18897, password="run-pw")
         d = self.runtime("a5")
-        self.seed_agent("a5", 18800,
+        self.seed_agent("a5", port,
-                        "--addr 127.0.0.1:18800 --data %s --plugins %s --auth auto"
+                        f"--addr 127.0.0.1:{port} --data %s --plugins %s --auth auto"
                         % (d / "data", d / "plugins"),
                         extra_rows=[("dbref", "pg1/agent"), ("db", "pg")])
         self.start_ok("a5")
@@ -437,10 +464,13 @@ class SvcDbRecoveryTests(unittest.TestCase):
     # ---- 6. 认证预检与后端口径一致（环境变量 DSN 覆盖 auto）----
 
     def test_auth_preflight_aligns_with_effective_env_dsn(self):
+        nl1_port = self.available_port()
+        nl2_port = self.available_port()
+        nl3_port = self.available_port()
         # 非回环 + auto + 无任何 DSN → 预检快速拒绝，不启动进程。
         d = self.runtime("nl1")
-        self.seed_agent("nl1", 18801,
+        self.seed_agent("nl1", nl1_port,
-                        "--addr 10.255.255.1:18801 --data %s --plugins %s --auth auto"
+                        f"--addr 10.255.255.1:{nl1_port} --data %s --plugins %s --auth auto"
                         % (d / "data", d / "plugins"))
         self.clear_serve_record()
         result = self.svc("nl1", "start", check=False)
@@ -451,9 +481,9 @@ class SvcDbRecoveryTests(unittest.TestCase):
         # 环境变量提供 DSN → auto 判为已开认证：预检放行（桩绑定非回环地址失败即退，
         # 区别于预检拒绝——证明进入启动路径且 DSN 经环境交付）。
         d = self.runtime("nl2")
-        self.seed_agent("nl2", 18802,
+        self.seed_agent("nl2", nl2_port,
-                        "--addr 10.255.255.1:18802 --data %s --plugins %s --auth auto"
+                        f"--addr 10.255.255.1:{nl2_port} --data %s --plugins %s --auth auto"
-                        % (d / "data", d / "plugins"), addr="10.255.255.1:18802")
+                        % (d / "data", d / "plugins"), addr=f"10.255.255.1:{nl2_port}")
         env_dsn = "postgresql://envu:env-pw@127.0.0.1:18898/agent"
         env2 = dict(self.env, SVC_NL2_DSN=env_dsn)
         self.clear_serve_record()
@@ -464,9 +494,9 @@ class SvcDbRecoveryTests(unittest.TestCase):
 
         # 显式 auth=off 覆盖环境 DSN → 仍拒绝。
         d = self.runtime("nl3")
-        self.seed_agent("nl3", 18803,
+        self.seed_agent("nl3", nl3_port,
-                        "--addr 10.255.255.1:18803 --data %s --plugins %s --auth off"
+                        f"--addr 10.255.255.1:{nl3_port} --data %s --plugins %s --auth off"
-                        % (d / "data", d / "plugins"), addr="10.255.255.1:18803")
+                        % (d / "data", d / "plugins"), addr=f"10.255.255.1:{nl3_port}")
         self.clear_serve_record()
         result = self.svc("nl3", "start", check=False,
                           env=dict(self.env, SVC_NL3_DSN=env_dsn))
@@ -477,10 +507,11 @@ class SvcDbRecoveryTests(unittest.TestCase):
     # ---- 7. 显式解绑：环境残留不再悄悄重新选库 ----
 
     def test_explicit_unbind_blocks_residual_env_dsn(self):
+        port = self.available_port()
         self.register_pg("pg1", 18899, password="unbind-pw")
         d = self.runtime("dev")
-        self.seed_agent("dev", 18804,
+        self.seed_agent("dev", port,
-                        "--addr 127.0.0.1:18804 --data %s --plugins %s"
+                        f"--addr 127.0.0.1:{port} --data %s --plugins %s"
                         % (d / "data", d / "plugins"))
         self.svc("dev", "db", "pg1", "--no-restart")
         self.assertEqual(self.kv("dev", "dbref"), "pg1/agent")
@@ -508,10 +539,12 @@ class SvcDbRecoveryTests(unittest.TestCase):
     # ---- 8. mock 冲突与旧产物：行为明确，不隐式降级 ----
 
     def test_mock_conflict_and_legacy_binary_refuse(self):
+        mock_port = self.available_port()
+        legacy_port = self.available_port()
         self.register_pg("pg1", 18902, password="mock-pw")
         d = self.runtime("mk1")
-        self.seed_agent("mk1", 18805,
+        self.seed_agent("mk1", mock_port,
-                        "--addr 127.0.0.1:18805 --data %s --plugins %s --auth auto"
+                        f"--addr 127.0.0.1:{mock_port} --data %s --plugins %s --auth auto"
                         % (d / "data", d / "plugins"),
                         extra_rows=[("dbref", "pg1/agent"), ("db", "pg")])
         result = self.svc("mk1", "deploy", "--from", "local", "--mock", check=False)
@@ -522,8 +555,8 @@ class SvcDbRecoveryTests(unittest.TestCase):
 
         # 旧产物（不消费环境变量 DSN）绑定服务 restart → 拒绝，提示可操作出路。
         d = self.runtime("lb1")
-        self.seed_agent("lb1", 18806,
+        self.seed_agent("lb1", legacy_port,
-                        "--addr 127.0.0.1:18806 --data %s --plugins %s --auth auto"
+                        f"--addr 127.0.0.1:{legacy_port} --data %s --plugins %s --auth auto"
                         % (d / "data", d / "plugins"),
                         extra_rows=[("dbref", "pg1/agent"), ("db", "pg")])
         env_legacy = dict(self.env, SVC_STUB_NO_ENV_DSN="1")
@@ -535,15 +568,17 @@ class SvcDbRecoveryTests(unittest.TestCase):
     # ---- 9. 多实例顺序更新不串用 DSN；凭据不进日志与历史 ----
 
     def test_sequential_restarts_no_cross_dsn_and_no_secret_leak(self):
+        dev_port = self.available_port()
+        prod_port = self.available_port()
         self.register_pg("pg1", 18903, password="iso-pw-one")
         self.register_pg("pg2", 18904, password="iso-pw-two")
         d1, d2 = self.runtime("dev"), self.runtime("prod")
-        self.seed_agent("dev", 18807,
+        self.seed_agent("dev", dev_port,
-                        "--addr 127.0.0.1:18807 --data %s --plugins %s --auth auto"
+                        f"--addr 127.0.0.1:{dev_port} --data %s --plugins %s --auth auto"
                         % (d1 / "data", d1 / "plugins"),
                         extra_rows=[("dbref", "pg1/agent"), ("db", "pg")])
-        self.seed_agent("prod", 18808,
+        self.seed_agent("prod", prod_port,
-                        "--addr 127.0.0.1:18808 --data %s --plugins %s --auth auto"
+                        f"--addr 127.0.0.1:{prod_port} --data %s --plugins %s --auth auto"
                         % (d2 / "data", d2 / "plugins"),
                         extra_rows=[("dbref", "pg2/agent"), ("db", "pg")])
         dsn1 = "postgresql://agent:iso-pw-one@127.0.0.1:18903/agent"
@@ -602,12 +637,13 @@ class SvcDbRecoveryTests(unittest.TestCase):
         return shas[1], shas[0]  # (旧, 新)
 
     def test_update_preserves_binding(self):
+        port = self.available_port()
         old_sha, new_sha = self.make_src_repo()
         self.register_pg("pg1", 18905, password="upd-pw")
         dsn = "postgresql://agent:upd-pw@127.0.0.1:18905/agent"
         d = self.runtime("dev")
-        self.seed_agent("dev", 18809,
+        self.seed_agent("dev", port,
-                        "--addr 127.0.0.1:18809 --data %s --plugins %s --auth auto"
+                        f"--addr 127.0.0.1:{port} --data %s --plugins %s --auth auto"
                         % (d / "data", d / "plugins"),
                         extra_rows=[("dbref", "pg1/agent"), ("db", "pg"),
                                     ("source", "dev@%s" % old_sha)])
@@ -630,7 +666,7 @@ class SvcDbRecoveryTests(unittest.TestCase):
         旧进程占口 → 报「非本服务」。env_config 须按实例清空 SVC_PIDCHECK 等。
         """
         old_sha, new_sha = self.make_src_repo()
-        ports = {"a": 18841, "b": 18842}
+        ports = {name: self.available_port() for name in ("a", "b")}
         old_pids = {}
         for name, port in ports.items():
             d = self.runtime(name)
@@ -694,11 +730,12 @@ class SvcDbRecoveryTests(unittest.TestCase):
         return env
 
     def test_seed_puts_with_temporary_bearer_key_when_auth_on(self):
+        port = self.available_port()
         self.register_pg("pg1", 18831, password="seed-pw-1")
         dsn = "postgresql://agent:seed-pw-1@127.0.0.1:18831/agent"
         d = self.runtime("dev")
-        self.seed_agent("dev", 18832,
+        self.seed_agent("dev", port,
-                        "--addr 127.0.0.1:18832 --data %s --plugins %s --auth auto --dsn %s"
+                        f"--addr 127.0.0.1:{port} --data %s --plugins %s --auth auto --dsn %s"
                         % (d / "data", d / "plugins", dsn))
         result = self.svc("dev", "start", env=self.model_env())
         self.assertIn("已播种设置：主模型", result.stdout)
@@ -714,11 +751,12 @@ class SvcDbRecoveryTests(unittest.TestCase):
         self.assertIn("apikey revoke --dsn %s --owner op --name svc-seed" % dsn, cli)
 
     def test_seed_owner_override_skips_account_lookup(self):
+        port = self.available_port()
         self.register_pg("pg1", 18833, password="seed-pw-2")
         dsn = "postgresql://agent:seed-pw-2@127.0.0.1:18833/agent"
         d = self.runtime("dev")
-        self.seed_agent("dev", 18834,
+        self.seed_agent("dev", port,
-                        "--addr 127.0.0.1:18834 --data %s --plugins %s --auth auto --dsn %s"
+                        f"--addr 127.0.0.1:{port} --data %s --plugins %s --auth auto --dsn %s"
                         % (d / "data", d / "plugins", dsn))
         result = self.svc("dev", "start", env=self.model_env(SVC_SEED_OWNER="boss"))
         self.assertIn("已播种设置：主模型", result.stdout)
@@ -728,9 +766,10 @@ class SvcDbRecoveryTests(unittest.TestCase):
         self.assertNotIn("user list", cli)  # 显式指定即跳过账号解析
 
     def test_seed_puts_without_credentials_when_auth_off(self):
+        port = self.available_port()
         d = self.runtime("dev")
-        self.seed_agent("dev", 18835,
+        self.seed_agent("dev", port,
-                        "--addr 127.0.0.1:18835 --data %s --plugins %s --auth auto"
+                        f"--addr 127.0.0.1:{port} --data %s --plugins %s --auth auto"
                         % (d / "data", d / "plugins"))
         result = self.svc("dev", "start", env=self.model_env())
         self.assertIn("已播种设置：主模型", result.stdout)
@@ -767,7 +806,7 @@ class SvcDbRecoveryTests(unittest.TestCase):
         run(work, "push", "-u", "origin", "main")
         return remote, work
 
-    def rdm_env(self, remote):
+    def rdm_env(self, remote, agent_name="rdm1"):
         env = dict(self.env)
         env.update({
             "SVC_RDM_SRC_DIR": str(self.root / "rdmcache"),
@@ -776,6 +815,8 @@ class SvcDbRecoveryTests(unittest.TestCase):
             "SVC_MODEL_KEY": "sk-stub-model",
             "SVC_MODEL_NAME": "stub-model",
         })
+        env["SVC_%sWEB_ADDR" % agent_name.upper()] = (
+            "127.0.0.1:%d" % self.available_port())
         return env
 
     def http_get(self, url, timeout=2):
@@ -796,9 +837,10 @@ class SvcDbRecoveryTests(unittest.TestCase):
         return "http://%s:%s/" % (host, port)
 
     def test_rdm_source_deploy_end_to_end(self):
+        port = self.available_port()
         remote, work = self.make_rdm_repo()
         env = self.rdm_env(remote)
-        result = self.svc("rdm1", "deploy", "--from", "rdm", "--port", "18841", env=env)
+        result = self.svc("rdm1", "deploy", "--from", "rdm", "--port", port, env=env)
         d = self.runtime("rdm1")
         # U2：组合产物入不可变 release，经 current 指针解析；插件空集、工作台随 release。
         active = d / "current"
@@ -822,9 +864,10 @@ class SvcDbRecoveryTests(unittest.TestCase):
         self.assertNotIn("user list", self.cli_record())
 
     def test_rdm_companion_missing_web_fails_deploy(self):
+        port = self.available_port()
         remote, work = self.make_rdm_repo(with_web=False)
-        env = self.rdm_env(remote)
+        env = self.rdm_env(remote, "rdm2")
-        result = self.svc("rdm2", "deploy", "--from", "rdm", "--port", "18843",
+        result = self.svc("rdm2", "deploy", "--from", "rdm", "--port", port,
                           env=env, check=False)
         self.assertNotEqual(result.returncode, 0)
         self.assertIn("ui-missing", result.stdout)
@@ -833,20 +876,21 @@ class SvcDbRecoveryTests(unittest.TestCase):
         self.assertIsNone(self.kv("rdm2web", "kind"))
 
     def test_source_switch_requires_flag(self):
+        port = self.available_port()
         remote, work = self.make_rdm_repo()
-        env = self.rdm_env(remote)
+        env = self.rdm_env(remote, "sw1")
         # 先以 local 源部署（内嵌 UI 桩一律 200）。
-        self.svc("sw1", "deploy", "--from", "local", "--port", "18844")
+        self.svc("sw1", "deploy", "--from", "local", "--port", port)
         old = self.kv("sw1", "source")
         self.assertTrue(old.startswith("local@"))
         # 无旗标换到 rdm → 拒绝，source 不变。
-        denied = self.svc("sw1", "deploy", "--from", "rdm", "--port", "18844",
+        denied = self.svc("sw1", "deploy", "--from", "rdm", "--port", port,
                           env=env, check=False)
         self.assertNotEqual(denied.returncode, 0)
         self.assertIn("allow-source-switch", denied.stdout)
         self.assertEqual(self.kv("sw1", "source"), old)
         # 带旗标允许换源，伴生就位。
-        ok = self.svc("sw1", "deploy", "--from", "rdm", "--port", "18844",
+        ok = self.svc("sw1", "deploy", "--from", "rdm", "--port", port,
                       "--allow-source-switch", env=env)
         self.assertTrue(self.kv("sw1", "source").startswith("rdm@"))
         self.assertIn("UI 就绪", ok.stdout)
@@ -855,6 +899,7 @@ class SvcDbRecoveryTests(unittest.TestCase):
         self.assertIn("rdm workbench", body)
 
     def test_rdm_source_commits_and_update_track(self):
+        port = self.available_port()
         remote, work = self.make_rdm_repo()
         env = self.rdm_env(remote)
         # commits rdm 列出来自独立 RDM 源仓的可部署提交。
@@ -862,7 +907,7 @@ class SvcDbRecoveryTests(unittest.TestCase):
         self.assertIn("RDM 仓提交", result.stdout)
         self.assertIn("rdm v1", result.stdout)
         # 部署 v1 → 远端推进 v2 → update 检测可更新并真更新（source 保持 rdm@ 轨道）。
-        self.svc("rdm1", "deploy", "--from", "rdm", "--port", "18842", env=env)
+        self.svc("rdm1", "deploy", "--from", "rdm", "--port", port, env=env)
         old = self.kv("rdm1", "source")
         (work / "marker").write_text("v2\n")
         subprocess.run(["git", "add", "."], cwd=work, env=self.env, timeout=60, check=True)
@@ -887,7 +932,7 @@ class SvcDbRecoveryTests(unittest.TestCase):
         htdocs = self.root / "static-site"
         htdocs.mkdir()
         (htdocs / "index.html").write_text("<html>static-ok</html>\n")
-        port = 18951
+        port = self.available_port()
         result = self.svc(
             "uiqa", "deploy", "--kind", "static",
             "--dir", str(htdocs), "--addr", "127.0.0.1:%d" % port)

tests/test_svc_recover.py

@@ -189,6 +189,7 @@ class SvcAutoRecoverTests(unittest.TestCase):
         tmp = tempfile.mkdtemp(prefix="svc-autorecover-")
         self.addCleanup(shutil.rmtree, tmp, True)
         self.root = Path(tmp)
+        self._allocated_ports = set()
         (self.root / "scripts").mkdir(parents=True)
         shutil.copy2(SVC_SCRIPT, self.root / "scripts" / "svc.sh")
         self.trace = self.root / "trace"
@@ -248,6 +249,21 @@ class SvcAutoRecoverTests(unittest.TestCase):
             self.assertEqual(proc.returncode, 0, proc.stdout)
         return proc
 
+    def available_port(self):
+        """Allocate a distinct service port for this test, including address changes.
+
+        The socket closes before svc.sh binds it, leaving a small allocation-to-start
+        race with unrelated processes. Keep allocated numbers distinct within the
+        test so stopped/restarted services never share another instance's port.
+        """
+        while True:
+            with socket.socket() as sock:
+                sock.bind(("0.0.0.0", 0))
+                port = sock.getsockname()[1]
+            if port not in self._allocated_ports:
+                self._allocated_ports.add(port)
+                return port
+
     def runtime(self, name):
         return self.root / "runtime" / name
 
@@ -330,14 +346,16 @@ class SvcAutoRecoverTests(unittest.TestCase):
     # ---- 1. 期望态：start 成功置位 / 显式 stop 清除（含进程已死的 stop）----
 
     def test_start_records_autostart_and_stop_clears(self):
-        self.seed_agent("dev", 18801)
+        port = self.available_port()
+        self.seed_agent("dev", port)
         self.start_ok("dev")
         self.assertEqual(self.kv("dev", "autostart"), "1")
         self.svc("dev", "stop")
         self.assertIsNone(self.kv("dev", "autostart"))
 
     def test_stop_on_dead_process_clears_autostart(self):
-        self.seed_agent("dev", 18802)
+        port = self.available_port()
+        self.seed_agent("dev", port)
         self.start_ok("dev")
         self.kill9("dev")
         # 进程已死但 autostart 保留（等待 recover）；显式 stop 表示"保持停止"。
@@ -348,7 +366,8 @@ class SvcAutoRecoverTests(unittest.TestCase):
     # ---- 2. recover：崩溃拉回 / 运行中不动 / 显式停止不复活 / 未启动不碰 ----
 
     def test_recover_restarts_sigkilled_service(self):
-        self.seed_agent("dev", 18803)
+        port = self.available_port()
+        self.seed_agent("dev", port)
         self.start_ok("dev")
         old_pid = self.kill9("dev")
         result = self.svc("recover", "--quiet")
@@ -361,10 +380,11 @@ class SvcAutoRecoverTests(unittest.TestCase):
         ok_events = self.history_events("dev", "recover")
         self.assertTrue(any('"result":"ok"' in l for l in ok_events), ok_events)
         # 恢复用的仍是持久化 exec=（svc.env 登记未漂移）。
-        self.assertIn("--addr 127.0.0.1:18803", self.kv("dev", "exec"))
+        self.assertIn(f"--addr 127.0.0.1:{port}", self.kv("dev", "exec"))
 
     def test_recover_noop_while_running(self):
-        self.seed_agent("dev", 18804)
+        port = self.available_port()
+        self.seed_agent("dev", port)
         self.start_ok("dev")
         pid = self.pid_of("dev")
         result = self.svc("recover")
@@ -373,7 +393,8 @@ class SvcAutoRecoverTests(unittest.TestCase):
         self.assertEqual(self.history_events("dev", "recover"), [])
 
     def test_recover_respects_explicit_stop(self):
-        self.seed_agent("dev", 18805)
+        port = self.available_port()
+        self.seed_agent("dev", port)
         self.start_ok("dev")
         self.svc("dev", "stop")
         result = self.svc("recover")
@@ -382,7 +403,8 @@ class SvcAutoRecoverTests(unittest.TestCase):
         self.assertEqual(self.history_events("dev", "recover"), [])
 
     def test_recover_ignores_never_started_service(self):
-        self.seed_agent("dev", 18806)  # 从未 start：无 autostart 键
+        port = self.available_port()
+        self.seed_agent("dev", port)  # 从未 start：无 autostart 键
         result = self.svc("recover")
         self.assertEqual(result.returncode, 0, result.stdout)
         self.assertFalse(self.pid_alive(self.pid_of("dev")))
@@ -391,7 +413,8 @@ class SvcAutoRecoverTests(unittest.TestCase):
     # ---- 3. 并发防护：生命周期锁被持有时 recover 跳过 ----
 
     def test_recover_skips_when_lock_held(self):
-        self.seed_agent("dev", 18807)
+        port = self.available_port()
+        self.seed_agent("dev", port)
         self.start_ok("dev")
         self.kill9("dev")
         lock = self.runtime("dev") / ".start.lock"
@@ -411,11 +434,12 @@ class SvcAutoRecoverTests(unittest.TestCase):
     # ---- 4. 恢复失败路径：端口被外部进程占用 → fail 事件 + 非零退出 ----
 
     def test_recover_fails_when_port_stolen(self):
-        self.seed_agent("dev", 18808)
+        port = self.available_port()
+        self.seed_agent("dev", port)
         self.start_ok("dev")
         self.kill9("dev")
         thief = subprocess.Popen(
-            ["python3", "-m", "http.server", "18808", "--bind", "127.0.0.1"],
+            ["python3", "-m", "http.server", str(port), "--bind", "127.0.0.1"],
             stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
         self.addCleanup(thief.terminate)
         # 等 thief 真正完成 bind 再 recover——慢环境里 Popen 返回时端口尚未监听，
@@ -426,7 +450,7 @@ class SvcAutoRecoverTests(unittest.TestCase):
             s = socket.socket()
             s.settimeout(0.2)
             try:
-                s.connect(("127.0.0.1", 18808))
+                s.connect(("127.0.0.1", port))
                 bound = True
             except OSError:
                 pass
@@ -435,7 +459,7 @@ class SvcAutoRecoverTests(unittest.TestCase):
             if bound:
                 break
             time.sleep(0.05)
-        self.assertTrue(bound, "thief 未能在 5s 内监听 18808")
+        self.assertTrue(bound, f"thief 未能在 5s 内监听 {port}")
         try:
             result = self.svc("recover", check=False)
             self.assertEqual(result.returncode, 1, result.stdout)
@@ -479,7 +503,7 @@ class SvcAutoRecoverTests(unittest.TestCase):
         """模拟 update 误清 pidfile 后本 bin 仍占口：裸 start 应回收并拉起新进程。"""
         if not self._ss_can_attribute_pid():
             self.skipTest("ss -ltnp 无法归因 PID（reclaim_own_port 需此能力）")
-        port = 18812
+        port = self.available_port()
         self.seed_agent("dev", port)
         self.start_ok("dev")
         old = self.pid_of("dev")
@@ -500,7 +524,7 @@ class SvcAutoRecoverTests(unittest.TestCase):
 
     def test_start_rejects_foreign_port_occupant(self):
         """外部 python http.server 占口：start 必须失败并提示非本服务。"""
-        port = 18813
+        port = self.available_port()
         self.seed_agent("dev", port)
         thief = subprocess.Popen(
             ["python3", "-m", "http.server", str(port), "--bind", "127.0.0.1"],
@@ -537,7 +561,8 @@ class SvcAutoRecoverTests(unittest.TestCase):
         self.assertFalse((unit_dir / "agent-system-svc-recover.service").exists())
 
     def test_recover_reserved_name_rejected(self):
-        self.seed_agent("recover", 18809)  # 保留字不可作服务名
+        port = self.available_port()
+        self.seed_agent("recover", port)  # 保留字不可作服务名
         result = self.svc("recover", "recover", check=False)
         self.assertEqual(result.returncode, 2, result.stdout)