agent-exec transport b7a8e78a2b6d0d8f6803490c7fa8af2e/9075eee055af

scripts/dhf_catalog_guard.py

@@ -0,0 +1,244 @@
+#!/usr/bin/env python3
+"""Compare the DHF inventory, immutable originals and one RDM asset catalog.
+
+This checks provenance and publication metadata. It cannot decide document
+meaning or Word rendering fidelity; those require the per-asset review.
+"""
+
+import argparse
+from collections import Counter
+from datetime import datetime
+import hashlib
+import json
+from pathlib import Path
+import re
+import subprocess
+import sys
+
+
+SOURCE_PREFIX = (".templates", "DHF 文件模板")
+STATES = {"needs-review", "blocked", "published", "retired"}
+SHA256 = re.compile(r"[0-9a-f]{64}\Z")
+COMMIT = re.compile(r"[0-9a-f]{40}\Z")
+RFC3339 = re.compile(r"\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(?:\.\d+)?(?:Z|[+-]\d{2}:\d{2})\Z")
+
+
+def digest(raw):
+    return hashlib.sha256(raw).hexdigest()
+
+
+def source_path(dev_root, source):
+    path = Path(source)
+    if (path.is_absolute() or path.parts[:2] != SOURCE_PREFIX
+            or ".." in path.parts or path.suffix.lower() not in {".doc", ".docx"}):
+        raise ValueError(f"unsafe or non-Word source path: {source}")
+    segment = dev_root
+    for part in path.parts:
+        segment = segment / part
+        if segment.is_symlink():
+            raise ValueError(f"source path symlink: {source}")
+    absolute = dev_root / path
+    if not absolute.resolve().is_relative_to(dev_root.resolve()) or absolute.is_symlink():
+        raise ValueError(f"source path leaves the Dev tree: {source}")
+    return absolute
+
+
+def package_digest(root, file_ref):
+    relative = file_ref.get("path")
+    expected = file_ref.get("sha256")
+    if not isinstance(relative, str) or not isinstance(expected, str) or not SHA256.fullmatch(expected):
+        raise ValueError("published asset file reference missing path or hash")
+    path = Path(relative)
+    if path.is_absolute() or ".." in path.parts or path == Path("."):
+        raise ValueError("published asset file reference is unsafe")
+    segment = root
+    for part in path.parts:
+        segment = segment / part
+        if segment.is_symlink():
+            raise ValueError("published asset file symlink")
+    absolute = root / path
+    if not absolute.resolve().is_relative_to(root.resolve()) or absolute.is_symlink() or not absolute.is_file():
+        raise ValueError("published asset file missing or escapes catalog")
+    if digest(absolute.read_bytes()) != expected:
+        raise ValueError("published asset file hash mismatch")
+    return expected
+
+
+def audit(dev_root, inventory, catalog, expected_count, require_published=False,
+          catalog_root=None, identity_baseline=None):
+    records = inventory.get("templates")
+    assets = catalog.get("assets")
+    if not isinstance(records, list) or not isinstance(assets, list):
+        raise ValueError("inventory/templates or catalog/assets is missing")
+    if len(records) != expected_count or len(assets) != expected_count:
+        raise ValueError(f"expected {expected_count} logical sources and assets")
+    by_path = {}
+    ids = set()
+    for asset in assets:
+        provenance = asset.get("sourceProvenance") or {}
+        path = provenance.get("path")
+        template_id = asset.get("templateId")
+        if path in by_path or not isinstance(path, str):
+            raise ValueError("duplicate or missing provenance path")
+        if template_id in ids or not isinstance(template_id, str) or not template_id:
+            raise ValueError("duplicate or missing logical template ID")
+        by_path[path] = asset
+        ids.add(template_id)
+    if identity_baseline is not None:
+        prior = identity_baseline.get("assets")
+        if not isinstance(prior, list) or len(prior) != expected_count:
+            raise ValueError("identity baseline is incomplete")
+        prior_ids = {}
+        for asset in prior:
+            if not isinstance(asset, dict):
+                raise ValueError("identity baseline has invalid asset")
+            path = (asset.get("sourceProvenance") or {}).get("path")
+            if path in prior_ids or not isinstance(path, str):
+                raise ValueError("identity baseline has duplicate source")
+            prior_ids[path] = asset.get("templateId")
+        for path, asset in by_path.items():
+            if prior_ids.get(path) != asset["templateId"]:
+                raise ValueError(f"template identity drift: {path}")
+    seen = set()
+    states = Counter()
+    hashes = set()
+    candidate_differences = []
+    for record in records:
+        relative = record.get("source")
+        if not isinstance(relative, str) or relative in seen or relative not in by_path:
+            raise ValueError(f"inventory source missing, duplicate or unmapped: {relative}")
+        seen.add(relative)
+        asset = by_path[relative]
+        expected = record.get("sourceSha256")
+        found = (asset.get("source") or {}).get("sha256")
+        if not isinstance(expected, str) or not SHA256.fullmatch(expected) or found != expected:
+            raise ValueError(f"catalog source hash mismatch: {relative}")
+        path = source_path(dev_root, relative)
+        if not path.is_file() or digest(path.read_bytes()) != expected:
+            raise ValueError(f"checkout hash mismatch: {relative}")
+        provenance = asset["sourceProvenance"]
+        commit = provenance.get("commit")
+        if provenance.get("repository") != "Agent-System-Dev" or not isinstance(commit, str) or not COMMIT.fullmatch(commit):
+            raise ValueError(f"untrusted source provenance: {relative}")
+        historical = subprocess.run(
+            ["git", "-C", str(dev_root), "show", f"{commit}:{relative}"],
+            capture_output=True, check=False,
+        )
+        if historical.returncode or digest(historical.stdout) != expected:
+            raise ValueError(f"committed source hash mismatch: {relative}")
+        if (asset.get("source") or {}).get("format") != path.suffix.lower()[1:]:
+            raise ValueError(f"catalog format mismatch: {relative}")
+        applicability = asset.get("applicability") or {}
+        catalog_types = applicability.get("docTypes", [])
+        if not isinstance(catalog_types, list):
+            raise ValueError(f"invalid catalog docTypes: {relative}")
+        if record.get("proposedDocType") not in catalog_types:
+            candidate_differences.append({
+                "path": relative,
+                "inventoryCandidate": record.get("proposedDocType"),
+                "catalogCandidates": catalog_types,
+                "confirmed": bool(applicability.get("confirmed")),
+            })
+        state = asset.get("state")
+        if state not in STATES:
+            raise ValueError(f"invalid asset state: {relative}")
+        if state == "published":
+            review = asset.get("review") or {}
+            normalized = asset.get("normalizedDocx") or {}
+            condition = applicability.get("condition")
+            if (not normalized or not catalog_types or applicability.get("confirmed") is not True
+                    or not isinstance(condition, str) or not condition.strip()
+                    or review.get("decision") != "approved"
+                    or review.get("sourceSha256") != expected
+                    or not asset.get("structure") or not asset.get("render")):
+                raise ValueError(f"published asset lacks approved binding: {relative}")
+            if (normalized.get("format") != "docx" or asset["structure"].get("format") != "json"
+                    or asset["render"].get("format") != "docx"):
+                raise ValueError(f"published asset has invalid derived formats: {relative}")
+            reviewer = review.get("reviewer")
+            evidence = review.get("evidence")
+            reviewed_at = review.get("reviewedAt")
+            if (not isinstance(reviewer, str) or not reviewer.strip()
+                    or not isinstance(evidence, str) or not evidence.strip()
+                    or not isinstance(reviewed_at, str)):
+                raise ValueError(f"published asset lacks reviewer evidence: {relative}")
+            if not RFC3339.fullmatch(reviewed_at):
+                raise ValueError(f"published asset has invalid review time: {relative}")
+            try:
+                at = datetime.fromisoformat(reviewed_at.replace("Z", "+00:00"))
+            except ValueError as exc:
+                raise ValueError(f"published asset has invalid review time: {relative}") from exc
+            if at.tzinfo is None:
+                raise ValueError(f"published asset has invalid review time: {relative}")
+            converter = asset.get("converter") or {}
+            if (asset["source"].get("format") == "doc" or
+                    asset["source"].get("sha256") != normalized.get("sha256")):
+                if (converter.get("inputSha256") != expected or
+                        converter.get("outputSha256") != normalized.get("sha256") or
+                        any(not isinstance(converter.get(field), str) or not converter[field].strip()
+                            for field in ("name", "version", "configuration"))):
+                    raise ValueError(f"published asset lacks conversion provenance: {relative}")
+            if any(issue.get("status") != "resolved" for issue in asset.get("issues", [])):
+                raise ValueError(f"published asset has unresolved issues: {relative}")
+            if (review.get("structureSha256") != asset["structure"].get("sha256")
+                    or review.get("renderSha256") != asset["render"].get("sha256")
+                    or review.get("normalizedSha256") != normalized.get("sha256")):
+                raise ValueError(f"published asset review hash mismatch: {relative}")
+            if catalog_root is not None:
+                package_digest(catalog_root, asset["source"])
+                package_digest(catalog_root, normalized)
+                package_digest(catalog_root, asset["structure"])
+                package_digest(catalog_root, asset["render"])
+        states[state] += 1
+        hashes.add(expected)
+    if require_published and states.get("published", 0) != expected_count:
+        raise ValueError(f"{expected_count - states.get('published', 0)} assets not published")
+    return {
+        "source": "SELF mechanical provenance check",
+        "matched": len(seen),
+        "logicalTemplateIds": len(ids),
+        "uniqueSourceBytes": len(hashes),
+        "published": states.get("published", 0),
+        "states": dict(sorted(states.items())),
+        "candidateTypeDifferences": candidate_differences,
+        "identityBaselineVerified": identity_baseline is not None,
+        "publicationValidation": "NOT_RUN: RDM LoadDHFAssetCatalog and human content/Word review are authoritative",
+        "limits": "This checks provenance and obvious metadata errors; state counts do not prove published bundles are valid.",
+    }
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--dev-root", type=Path, required=True)
+    parser.add_argument("--inventory", type=Path, required=True)
+    parser.add_argument("--catalog", type=Path, required=True)
+    parser.add_argument("--expected-count", type=int, default=44)
+    parser.add_argument("--identity-baseline", type=Path)
+    parser.add_argument("--require-published", action="store_true")
+    args = parser.parse_args()
+    try:
+        if args.expected_count < 1:
+            raise ValueError("expected-count must be positive")
+        inventory_raw = args.inventory.read_bytes()
+        catalog_raw = args.catalog.read_bytes()
+        baseline_raw = args.identity_baseline.read_bytes() if args.identity_baseline else None
+        result = audit(args.dev_root, json.loads(inventory_raw), json.loads(catalog_raw),
+                       args.expected_count, args.require_published, args.catalog.parent,
+                       json.loads(baseline_raw) if baseline_raw is not None else None)
+        if args.catalog.read_bytes() != catalog_raw:
+            raise ValueError("catalog changed during check")
+        result["catalogSha256"] = digest(catalog_raw)
+        result["inventorySha256"] = digest(inventory_raw)
+        if baseline_raw is not None:
+            result["identityBaselineSha256"] = digest(baseline_raw)
+        if args.require_published:
+            raise ValueError("RDM publication validator required for full bundle safety and human fidelity")
+    except (OSError, ValueError, json.JSONDecodeError) as exc:
+        print(f"dhf catalog guard: {exc}", file=sys.stderr)
+        return 1
+    print(json.dumps(result, ensure_ascii=False, sort_keys=True))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

tests/test_dhf_catalog_guard.py

@@ -0,0 +1,256 @@
+import hashlib
+import json
+from pathlib import Path
+import subprocess
+import sys
+import tempfile
+import unittest
+from copy import deepcopy
+
+
+ROOT = Path(__file__).resolve().parents[1]
+GUARD = ROOT / "scripts" / "dhf_catalog_guard.py"
+
+
+class DHFCatalogGuardTest(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.addCleanup(self.tmp.cleanup)
+        self.root = Path(self.tmp.name)
+        self.dev = self.root / "dev"
+        self.dev.mkdir()
+        subprocess.run(["git", "init", "-q", str(self.dev)], check=True)
+        self.items = []
+        self.assets = []
+        for index, value in enumerate((b"source-one", b"source-two"), 1):
+            path = f".templates/DHF 文件模板/阶段{index}/模板{index}.docx"
+            absolute = self.dev / path
+            absolute.parent.mkdir(parents=True)
+            absolute.write_bytes(value)
+            digest = hashlib.sha256(value).hexdigest()
+            self.items.append({
+                "source": path,
+                "sourceSha256": digest,
+                "proposedDocType": f"type-{index}",
+            })
+            self.assets.append({
+                "templateId": f"dhf.logical-{index}",
+                "version": "1.0.0",
+                "state": "needs-review",
+                "source": {"sha256": digest, "format": "docx"},
+                "sourceProvenance": {"repository": "Agent-System-Dev", "path": path},
+                "applicability": {"docTypes": [f"type-{index}"], "confirmed": False},
+            })
+        subprocess.run(["git", "-C", str(self.dev), "add", ".templates"], check=True)
+        subprocess.run([
+            "git", "-C", str(self.dev), "-c", "user.name=Test",
+            "-c", "user.email=test@example.invalid", "commit", "-qm", "sources",
+        ], check=True)
+        commit = subprocess.check_output(
+            ["git", "-C", str(self.dev), "rev-parse", "HEAD"], text=True,
+        ).strip()
+        for asset in self.assets:
+            asset["sourceProvenance"]["commit"] = commit
+        self.inventory = self.root / "inventory.json"
+        self.catalog = self.root / "catalog.json"
+        self.save()
+
+    def save(self):
+        self.inventory.write_text(json.dumps({"templates": self.items}))
+        self.catalog.write_text(json.dumps({"schemaVersion": "dhf-assets/v1", "assets": self.assets}))
+
+    def run_guard(self, *extra):
+        return subprocess.run([
+            sys.executable, str(GUARD), "--dev-root", str(self.dev),
+            "--inventory", str(self.inventory), "--catalog", str(self.catalog),
+            "--expected-count", "2", *extra,
+        ], capture_output=True, text=True)
+
+    def publish_fixture(self, index=0, *, structure=b'{"blocks":[]}', render=b"rendered-docx"):
+        asset = self.assets[index]
+        number = index + 1
+        source = (self.dev / self.items[index]["source"]).read_bytes()
+        names = {"source": f"source{number}.docx", "structure": f"structure{number}.json",
+                 "render": f"render{number}.docx"}
+        for label, raw in [("source", source), ("structure", structure), ("render", render)]:
+            (self.root / names[label]).write_bytes(raw)
+        asset["state"] = "published"
+        asset["applicability"].update({"confirmed": True, "condition": "test scope"})
+        asset["source"]["path"] = names["source"]
+        asset["normalizedDocx"] = {"path": names["source"], "format": "docx",
+                                   "sha256": hashlib.sha256(source).hexdigest()}
+        asset["structure"] = {"path": names["structure"], "format": "json",
+                              "sha256": hashlib.sha256(structure).hexdigest()}
+        asset["render"] = {"path": names["render"], "format": "docx",
+                           "sha256": hashlib.sha256(render).hexdigest()}
+        asset["review"] = {
+            "decision": "approved", "reviewer": "reviewer", "reviewedAt": "2026-10-14T00:00:00Z",
+            "evidence": "review-record", "sourceSha256": self.items[index]["sourceSha256"],
+            "normalizedSha256": asset["normalizedDocx"]["sha256"],
+            "structureSha256": asset["structure"]["sha256"],
+            "renderSha256": asset["render"]["sha256"],
+        }
+        return asset
+
+    def test_accepts_matching_immutable_sources_but_does_not_call_them_published(self):
+        result = self.run_guard()
+        self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
+        report = json.loads(result.stdout)
+        self.assertEqual(report["matched"], 2)
+        self.assertEqual(report["states"], {"needs-review": 2})
+        self.assertEqual(report["published"], 0)
+
+    def test_rejects_changed_checkout_even_when_catalog_matches_historical_commit(self):
+        (self.dev / self.items[0]["source"]).write_bytes(b"changed")
+        result = self.run_guard()
+        self.assertNotEqual(result.returncode, 0)
+        self.assertIn("checkout hash mismatch", result.stderr)
+
+    def test_rejects_original_in_symlinked_parent_directory(self):
+        source = self.dev / self.items[0]["source"]
+        original_parent = source.parent
+        moved_parent = original_parent.parent / "alias-target"
+        original_parent.rename(moved_parent)
+        original_parent.symlink_to(moved_parent, target_is_directory=True)
+        result = self.run_guard()
+        self.assertNotEqual(result.returncode, 0)
+        self.assertIn("source path symlink", result.stderr)
+
+    def test_rejects_false_publication(self):
+        self.assets[0]["state"] = "published"
+        self.save()
+        result = self.run_guard()
+        self.assertNotEqual(result.returncode, 0)
+        self.assertIn("published asset lacks", result.stderr)
+
+    def test_reports_revised_candidate_type_without_approving_publication(self):
+        self.assets[0]["applicability"]["docTypes"] = ["unrelated"]
+        self.save()
+        result = self.run_guard()
+        self.assertEqual(result.returncode, 0, result.stderr)
+        report = json.loads(result.stdout)
+        self.assertEqual(report["published"], 0)
+        self.assertEqual(report["candidateTypeDifferences"], [{
+            "path": self.items[0]["source"],
+            "inventoryCandidate": "type-1",
+            "catalogCandidates": ["unrelated"],
+            "confirmed": False,
+        }])
+
+    def test_publication_gate_does_not_accept_a_complete_inventory_as_delivery(self):
+        result = self.run_guard("--require-published")
+        self.assertNotEqual(result.returncode, 0)
+        self.assertIn("not published", result.stderr)
+
+    def test_rejects_published_asset_whose_review_does_not_bind_bundle_bytes(self):
+        asset = self.publish_fixture()
+        asset["review"]["structureSha256"] = "0" * 64
+        self.save()
+        result = self.run_guard()
+        self.assertNotEqual(result.returncode, 0)
+        self.assertIn("published asset review hash mismatch", result.stderr)
+
+    def test_rejects_bundle_file_changed_after_approval(self):
+        asset = self.publish_fixture()
+        self.save()
+        (self.root / asset["render"]["path"]).write_bytes(b"modified-after-review")
+        result = self.run_guard()
+        self.assertNotEqual(result.returncode, 0)
+        self.assertIn("published asset file hash mismatch", result.stderr)
+
+    def test_rejects_published_asset_with_unresolved_fidelity_issue(self):
+        asset = self.publish_fixture()
+        asset["issues"] = [{"id": "fidelity", "status": "open"}]
+        self.save()
+        result = self.run_guard()
+        self.assertNotEqual(result.returncode, 0)
+        self.assertIn("published asset has unresolved issues", result.stderr)
+
+    def test_publication_flag_never_claims_authoritative_rdm_validation(self):
+        for index in range(2):
+            self.publish_fixture(index, structure=b"not-a-valid-structure", render=b"not-a-valid-docx")
+        self.save()
+        result = self.run_guard("--require-published")
+        self.assertNotEqual(result.returncode, 0)
+        self.assertIn("RDM publication validator required", result.stderr)
+
+    def test_rejects_swapped_logical_ids_against_fixed_identity_baseline(self):
+        baseline = self.root / "identity-baseline.json"
+        baseline.write_text(json.dumps({"assets": deepcopy(self.assets)}))
+        self.assets[0]["templateId"], self.assets[1]["templateId"] = (
+            self.assets[1]["templateId"], self.assets[0]["templateId"])
+        self.save()
+        result = self.run_guard("--identity-baseline", str(baseline))
+        self.assertNotEqual(result.returncode, 0)
+        self.assertIn("template identity drift", result.stderr)
+
+    def test_report_binds_exact_catalog_bytes(self):
+        result = self.run_guard()
+        self.assertEqual(result.returncode, 0, result.stderr)
+        self.assertEqual(json.loads(result.stdout)["catalogSha256"],
+                         hashlib.sha256(self.catalog.read_bytes()).hexdigest())
+
+    def test_rejects_published_package_through_symlink_directory(self):
+        asset = self.publish_fixture()
+        real = self.root / "real"
+        real.mkdir()
+        for key in ["source", "normalizedDocx", "structure", "render"]:
+            p = self.root / asset[key]["path"]
+            (real / p.name).write_bytes(p.read_bytes())
+            asset[key]["path"] = "alias/" + p.name
+        (self.root / "alias").symlink_to(real, target_is_directory=True)
+        self.save()
+        result = self.run_guard()
+        self.assertNotEqual(result.returncode, 0)
+        self.assertIn("symlink", result.stderr)
+
+    def test_rejects_string_false_and_null_applicability(self):
+        asset = self.publish_fixture()
+        asset["applicability"]["confirmed"] = "false"
+        self.save()
+        result = self.run_guard()
+        self.assertNotEqual(result.returncode, 0)
+        self.assertIn("published asset lacks approved binding", result.stderr)
+
+        asset["applicability"]["confirmed"] = True
+        asset["applicability"]["condition"] = None
+        self.save()
+        result = self.run_guard()
+        self.assertNotEqual(result.returncode, 0)
+        self.assertIn("published asset lacks approved binding", result.stderr)
+
+    def test_rejects_null_reviewer_and_non_rfc3339_timestamp(self):
+        asset = self.publish_fixture()
+        asset["review"]["reviewer"] = None
+        self.save()
+        result = self.run_guard()
+        self.assertNotEqual(result.returncode, 0)
+        self.assertIn("published asset lacks reviewer evidence", result.stderr)
+
+        asset["review"]["reviewer"] = "reviewer"
+        asset["review"]["reviewedAt"] = "2026-10-14 00:00:00Z"
+        self.save()
+        result = self.run_guard()
+        self.assertNotEqual(result.returncode, 0)
+        self.assertIn("published asset has invalid review time", result.stderr)
+
+    def test_rejects_changed_normalized_file_without_named_converter(self):
+        asset = self.publish_fixture()
+        normalized = b"changed-normalized-docx"
+        (self.root / "normalized.docx").write_bytes(normalized)
+        asset["normalizedDocx"] = {
+            "path": "normalized.docx", "format": "docx", "sha256": hashlib.sha256(normalized).hexdigest(),
+        }
+        asset["review"]["normalizedSha256"] = asset["normalizedDocx"]["sha256"]
+        asset["converter"] = {
+            "inputSha256": self.items[0]["sourceSha256"],
+            "outputSha256": asset["normalizedDocx"]["sha256"],
+        }
+        self.save()
+        result = self.run_guard()
+        self.assertNotEqual(result.returncode, 0)
+        self.assertIn("published asset lacks conversion provenance", result.stderr)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_svc.py

@@ -72,7 +72,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" "$data" <<'PYSRV' >/dev/null 2>&1 &
 import json, os, sqlite3, sys, threading, time
 from http.server import BaseHTTPRequestHandler, HTTPServer
@@ -166,6 +166,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"
@@ -223,6 +224,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
 
@@ -292,11 +308,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()
@@ -318,11 +335,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()
@@ -342,30 +360,30 @@ 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")
-        with socket.socket() as reservation:
-            reservation.bind(("127.0.0.1", 0))
-            port = reservation.getsockname()[1]
         self.seed_agent("dev", port,
                         "--addr 127.0.0.1:%d --data %s --plugins %s --auth auto"
                         % (port, d / "data", d / "plugins"))
@@ -405,9 +423,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()
@@ -420,8 +442,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)
@@ -431,8 +453,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")])
@@ -442,8 +464,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)
@@ -452,10 +474,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")
@@ -473,10 +496,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)
@@ -487,9 +513,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()
@@ -500,9 +526,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))
@@ -513,10 +539,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")
@@ -544,10 +571,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)
@@ -558,8 +587,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")
@@ -571,15 +600,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"
@@ -638,12 +669,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)])
@@ -666,7 +698,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)
@@ -730,11 +762,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)
@@ -750,11 +783,12 @@ class SvcDbRecoveryTests(unittest.TestCase):
         self.assertIn("apikey revoke --dsn %s --owner op --name svc-seed" % dsn, cli)
 
     def test_seed_writes_fallbacks_and_vision(self):
+        port = self.available_port()
         self.register_pg("pg1", 18951, password="seed-pw-v")
         dsn = "postgresql://agent:seed-pw-v@127.0.0.1:18951/agent"
         d = self.runtime("dev")
-        self.seed_agent("dev", 18952,
+        self.seed_agent("dev", port,
-                        "--addr 127.0.0.1:18952 --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_MODEL_FALLBACKS="ZHIPU/GLM-5.3-FlashX,ZHIPU/GLM-5.3-Flash",
@@ -770,11 +804,12 @@ class SvcDbRecoveryTests(unittest.TestCase):
         self.assertIn('"capable": "deepseek-v4.1-flash,qwen3.8-flash"', body_part)
 
     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)
@@ -784,9 +819,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)
@@ -809,9 +845,10 @@ class SvcDbRecoveryTests(unittest.TestCase):
         return db
 
     def test_seed_replaces_old_system_secret(self):
+        port = self.available_port()
         d = self.runtime("dev")
-        self.seed_agent("dev", 18836,
+        self.seed_agent("dev", port,
-                        "--addr 127.0.0.1:18836 --data %s --plugins %s --auth auto"
+                        f"--addr 127.0.0.1:{port} --data %s --plugins %s --auth auto"
                         % (d / "data", d / "plugins"))
         db = self.plant_old_chat_secret("dev", "sk-old-secret")
         result = self.svc("dev", "start", env=self.model_env())
@@ -826,9 +863,10 @@ class SvcDbRecoveryTests(unittest.TestCase):
         self.assertEqual(active[0], "sk-stub-model")
 
     def test_seed_warns_when_old_secret_stays_active(self):
+        port = self.available_port()
         d = self.runtime("dev")
-        self.seed_agent("dev", 18837,
+        self.seed_agent("dev", port,
-                        "--addr 127.0.0.1:18837 --data %s --plugins %s --auth auto"
+                        f"--addr 127.0.0.1:{port} --data %s --plugins %s --auth auto"
                         % (d / "data", d / "plugins"))
         self.plant_old_chat_secret("dev", "sk-old-secret")
         result = self.svc("dev", "start", env=self.model_env(SVC_STUB_KEEP_OLD_SECRET="1"))
@@ -866,7 +904,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"),
@@ -875,6 +913,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):
@@ -895,9 +935,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", "18843", env=env)
+        result = self.svc("rdm1", "deploy", "--from", "rdm", "--port", port, env=env)
         d = self.runtime("rdm1")
         # U2：组合产物入不可变 release，经 current 指针解析；插件空集、工作台随 release。
         active = d / "current"
@@ -921,9 +962,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)
@@ -932,20 +974,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)
@@ -954,6 +997,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 源仓的可部署提交。
@@ -961,7 +1005,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)
@@ -986,7 +1030,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_realcombo.py

@@ -165,7 +165,8 @@ class RealComboTests(tsr.SvcAutoRecoverTests):
     def test_U3_real_agent_identity_lifecycle_sqlite(self):
         """真实组合二进制：领养 ensure → verify → 两次裸重启身份/锚点稳定；
         capabilities v1 探测契约。"""
-        self.seed_real("rc1", 9471, self.real_bin)
+        port = self.available_port()
+        self.seed_real("rc1", port, self.real_bin)
         result = self.svc("rc1", "start")
         self.assertIn("已启动", result.stdout)
         m = self.manifest("rc1")
@@ -178,7 +179,7 @@ class RealComboTests(tsr.SvcAutoRecoverTests):
         self.assertTrue(con.execute(
             "SELECT count(*) FROM schema_migrations").fetchone()[0] >= 1)
         con.close()
-        self.assertTrue(self.http_ok(9471))
+        self.assertTrue(self.http_ok(port))
 
         # capabilities v1 契约（探测无副作用形态）
         cap = subprocess.run([str(self.runtime("rc1") / "bin" / "agent-system"),
@@ -203,7 +204,7 @@ class RealComboTests(tsr.SvcAutoRecoverTests):
         self.assertEqual(self.identity_row("rc1"), [(iid, "sqlite")],
                          "裸重启不得换身份")
         self.assertEqual(self.manifest("rc1")["instance_id"], iid)
-        self.assertTrue(self.http_ok(9471))
+        self.assertTrue(self.http_ok(port))
         self.svc("rc1", "stop")
 
     @unittest.skipUnless(REAL_RDM_BIN, "真实 RDM 专项 NOT_RUN：需 SVC_REAL_RDM_BIN")
@@ -211,13 +212,14 @@ class RealComboTests(tsr.SvcAutoRecoverTests):
         """真实 RDM 组合二进制：identity 接线（Backend 公共 CLI）→ serve 门禁 →
         领域层可初始化。"""
         rdm_bin = Path(REAL_RDM_BIN).resolve()
-        self.seed_real("rcr", 9473, rdm_bin)
+        port = self.available_port()
+        self.seed_real("rcr", port, rdm_bin)
         result = self.svc("rcr", "start")
         self.assertIn("已启动", result.stdout)
         m = self.manifest("rcr")
         iid = m["instance_id"]
         self.assertEqual(self.identity_row("rcr"), [(iid, "sqlite")])
-        self.assertTrue(self.http_ok(9473))
+        self.assertTrue(self.http_ok(port))
         # RDM 未知子命令明确退出（不落服务路径）
         bogus = subprocess.run([str(self.runtime("rcr") / "bin" / "agent-system"),
                                 "totally-bogus"],
@@ -234,9 +236,11 @@ class RealComboTests(tsr.SvcAutoRecoverTests):
         self.env["RDM_DISABLE_EXEC_TOOLS"] = "1"
         env = dict(self.env)
         env.update(ci_git_env())
+        agent_port = self.available_port()
         env.update({"SVC_RDM_SRC_DIR": str(self.root / "rdmcache"),
-                    "SVC_RDM_ORIGIN": RDM_ORIGIN})
+                    "SVC_RDM_ORIGIN": RDM_ORIGIN,
-        result = self.svc("rcf", "deploy", "--from", "rdm", "--port", "9474",
+                    "SVC_RCFWEB_ADDR": "127.0.0.1:%d" % self.available_port()})
+        result = self.svc("rcf", "deploy", "--from", "rdm", "--port", agent_port,
                           env=env, check=False)
         if result.returncode != 0:
             log = self.runtime("rcf") / "svc.log"
@@ -259,7 +263,7 @@ class RealComboTests(tsr.SvcAutoRecoverTests):
                       else "rdm", "工作台内容")
         self.svc("rcf", "restart")
         self.assertEqual(self.identity_row("rcf"), [(iid, "sqlite")])
-        self.assertTrue(self.http_ok(9474))
+        self.assertTrue(self.http_ok(agent_port))
         self.svc("rcf", "stop")
         self.svc("rcfweb", "stop")
 

tests/test_svc_recover.py

@@ -184,17 +184,12 @@ esac
 """
 
 
-def free_fixture_port():
-    with socket.socket() as reservation:
-        reservation.bind(("127.0.0.1", 0))
-        return reservation.getsockname()[1]
-
-
 class SvcAutoRecoverTests(unittest.TestCase):
     def setUp(self):
         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"
@@ -254,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
 
@@ -336,14 +346,14 @@ class SvcAutoRecoverTests(unittest.TestCase):
     # ---- 1. 期望态：start 成功置位 / 显式 stop 清除（含进程已死的 stop）----
 
     def test_start_records_autostart_and_stop_clears(self):
-        self.seed_agent("dev", free_fixture_port())
+        self.seed_agent("dev", self.available_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", free_fixture_port())
+        self.seed_agent("dev", self.available_port())
         self.start_ok("dev")
         self.kill9("dev")
         # 进程已死但 autostart 保留（等待 recover）；显式 stop 表示"保持停止"。
@@ -354,7 +364,7 @@ class SvcAutoRecoverTests(unittest.TestCase):
     # ---- 2. recover：崩溃拉回 / 运行中不动 / 显式停止不复活 / 未启动不碰 ----
 
     def test_recover_restarts_sigkilled_service(self):
-        port = free_fixture_port()
+        port = self.available_port()
         self.seed_agent("dev", port)
         self.start_ok("dev")
         old_pid = self.kill9("dev")
@@ -371,7 +381,7 @@ class SvcAutoRecoverTests(unittest.TestCase):
         self.assertIn("--addr 127.0.0.1:%d" % port, self.kv("dev", "exec"))
 
     def test_recover_noop_while_running(self):
-        self.seed_agent("dev", free_fixture_port())
+        self.seed_agent("dev", self.available_port())
         self.start_ok("dev")
         pid = self.pid_of("dev")
         result = self.svc("recover")
@@ -380,7 +390,7 @@ class SvcAutoRecoverTests(unittest.TestCase):
         self.assertEqual(self.history_events("dev", "recover"), [])
 
     def test_recover_respects_explicit_stop(self):
-        self.seed_agent("dev", free_fixture_port())
+        self.seed_agent("dev", self.available_port())
         self.start_ok("dev")
         self.svc("dev", "stop")
         result = self.svc("recover")
@@ -389,7 +399,7 @@ class SvcAutoRecoverTests(unittest.TestCase):
         self.assertEqual(self.history_events("dev", "recover"), [])
 
     def test_recover_ignores_never_started_service(self):
-        self.seed_agent("dev", free_fixture_port())  # 从未 start：无 autostart 键
+        self.seed_agent("dev", self.available_port())  # 从未 start：无 autostart 键
         result = self.svc("recover")
         self.assertEqual(result.returncode, 0, result.stdout)
         self.assertFalse(self.pid_alive(self.pid_of("dev")))
@@ -398,7 +408,7 @@ class SvcAutoRecoverTests(unittest.TestCase):
     # ---- 3. 并发防护：生命周期锁被持有时 recover 跳过 ----
 
     def test_recover_skips_when_lock_held(self):
-        self.seed_agent("dev", free_fixture_port())
+        self.seed_agent("dev", self.available_port())
         self.start_ok("dev")
         self.kill9("dev")
         lock = self.runtime("dev") / ".start.lock"
@@ -418,7 +428,7 @@ class SvcAutoRecoverTests(unittest.TestCase):
     # ---- 4. 恢复失败路径：端口被外部进程占用 → fail 事件 + 非零退出 ----
 
     def test_recover_fails_when_port_stolen(self):
-        port = free_fixture_port()
+        port = self.available_port()
         self.seed_agent("dev", port)
         self.start_ok("dev")
         self.kill9("dev")
@@ -443,7 +453,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)
@@ -487,7 +497,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 = free_fixture_port()
+        port = self.available_port()
         self.seed_agent("dev", port)
         self.start_ok("dev")
         old = self.pid_of("dev")
@@ -508,7 +518,7 @@ class SvcAutoRecoverTests(unittest.TestCase):
 
     def test_start_rejects_foreign_port_occupant(self):
         """外部 python http.server 占口：start 必须失败并提示非本服务。"""
-        port = free_fixture_port()
+        port = self.available_port()
         self.seed_agent("dev", port)
         thief = subprocess.Popen(
             ["python3", "-m", "http.server", str(port), "--bind", "127.0.0.1"],
@@ -545,7 +555,7 @@ 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", free_fixture_port())  # 保留字不可作服务名
+        self.seed_agent("recover", self.available_port())  # 保留字不可作服务名
         result = self.svc("recover", "recover", check=False)
         self.assertEqual(result.returncode, 2, result.stdout)