U-036: real unittest coverage for supervisor/isolation + Deploy Supervisor skeleton

Ports graphloop's test_isolated_coding_workspace.py/test_workspace_security.py/ test_change_manifest.py/test_selective_apply.py into supervisor/isolation/ (only the import path changed -- the graphlemon modules were already adapted verbatim from graphloop, confirmed API-identical) plus a few extra edge cases the originals didn't cover. 52 tests green, replacing what had been a single manual smoke test since these modules were staged. Adds supervisor/deploy_supervisor.py: process lifecycle only (PID file, atomic status file, start/stop/is_running), no deploy decision logic yet -- that's U-037+'s job, injected via the poll_fn hook run_foreground() calls on a schedule. PID/status conventions adapted (not imported) from graphloop's hermes_watchdog.py atomic-write pattern. Fail-closed: any exit path (normal, SIGTERM, or an exception from poll_fn) unconditionally writes status='stopped' and removes the PID file, so is_running() never lies. RealSubprocessLifecycleTest actually spawns and stops a real `python3 -m supervisor.deploy_supervisor` process rather than mocking it, per this Unit's own "能独立启停" Done criterion. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

U-036: real unittest coverage for supervisor/isolation + Deploy Supervisor skeleton
Ports graphloop's test_isolated_coding_workspace.py/test_workspace_security.py/ test_change_manifest.py/test_selective_apply.py into supervisor/isolation/ (only the import path changed -- the graphlemon modules were already adapted verbatim from graphloop, confirmed API-identical) plus a few extra edge cases the originals didn't cover. 52 tests green, replacing what had been a single manual smoke test since these modules were staged. Adds supervisor/deploy_supervisor.py: process lifecycle only (PID file, atomic status file, start/stop/is_running), no deploy decision logic yet -- that's U-037+'s job, injected via the poll_fn hook run_foreground() calls on a schedule. PID/status conventions adapted (not imported) from graphloop's hermes_watchdog.py atomic-write pattern. Fail-closed: any exit path (normal, SIGTERM, or an exception from poll_fn) unconditionally writes status='stopped' and removes the PID file, so is_running() never lies. RealSubprocessLifecycleTest actually spawns and stops a real `python3 -m supervisor.deploy_supervisor` process rather than mocking it, per this Unit's own "能独立启停" Done criterion. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
bboonnee
1 parent 753fefa4
Showing 9 changed files with 1111 additions and 1 deletions
docs/development-plan.md
docs/sdd/unit_matrix.json
supervisor/__init__.py
supervisor/deploy_supervisor.py
supervisor/isolation/test_change_manifest.py
supervisor/isolation/test_isolated_workspace.py
supervisor/isolation/test_selective_apply.py
supervisor/isolation/test_workspace_security.py
supervisor/test_deploy_supervisor.py
@@ -354,6 +354,9 @@ PH2-S1/S2/S3/S4 之间除标注的依赖外相互独立，可并行推进。
   - 依赖：无（代码已存在，是补测试和补骨架，不是从零开发）
   - Done：四模块各自有对应 `test_*.py`（可直接照搬/改造 graphloop 的 `test_change_manifest.py`/`test_selective_apply.py`/`test_workspace_security.py` 等，逻辑未改动，测试也基本不用改）；Deploy Supervisor 骨架能独立启停，不受 Live/Candidate 进程崩溃影响
   - **待定（2026-07-31 订正）**：此前这里写的是"`supervisor/` 目录该归入 U-002 的目标项目树（`.claude/`、`skills/`、`src/` 那套）还是 U-003 的平台目录"——但 U-002 的范围已经订正为只创建 `.graphlemon/`（见该 Unit 2026-07-31 订正说明），`.claude/`、`skills/` 从来就不是目标项目的东西，是 graphlemon 平台自己仓库的资产。所以这个选择题已经不成立，`supervisor/` 只可能归属"platform 自身工具"（当前的实际存放位置，`supervisor/isolation/`），不存在真正的"归入目标项目"选项；保留这条记录是为了让后来者理解为什么这里曾经有过一次误判
+  - **实现补充（2026-08-02）**：四模块的单测直接照搬 graphloop 对应文件（`test_isolated_coding_workspace.py`/`test_workspace_security.py`/`test_change_manifest.py`/`test_selective_apply.py`），逐条核对后确认 API 与行为完全匹配（graphlemon 这四个模块本来就是"adapted verbatim"），只改了 import 路径（`from supervisor.isolation import X`）+ 补了几条 graphlemon 自己模块文档里提到但原测试没覆盖的小 case（`isolated_workspace` 的空/不存在 source_dir 拒绝、`workspace_security` 的换行符注入拒绝）；新增 `supervisor/__init__.py`（原本靠 Python 3 namespace package 也能跑，但补一个空的显式 `__init__.py` 和 `src/__init__.py` 保持同一约定，不依赖隐式行为）。52 个测试全绿，放在 `supervisor/isolation/test_*.py`（和被测模块同目录，比照 `src/test_*.py` 的既有约定），运行方式 `python3 -m unittest discover -s supervisor/isolation -p "test_*.py"`（不在 `python3 -m unittest discover -s src` 覆盖范围内，是新增的第二条独立测试入口，README/CLAUDE.md 后续需要记一笔）。
+    Deploy Supervisor 骨架落在新文件 `supervisor/deploy_supervisor.py`：PID 文件+状态文件走 `paths.platform_home()/deploy_supervisor/` 下（复用 U-003 已有的运行时目录约定，不新建一套），原子写沿用 `tempfile+os.replace`（参照 graphloop `hermes_watchdog.py` 的 `_atomic_save`，未直接 import，逻辑重写了一遍）。`run_foreground(poll_fn, poll_interval, iterations)` 是骨架的核心循环——`poll_fn` 是 U-037+ 才会真正实现的"检查是否有待部署的 commit 并行动"逻辑，本 Unit 只负责按周期调用它、写 PID/状态文件、处理 `SIGTERM`（收到后跳出循环，`finally` 块无条件写 `status='stopped'` 并删 PID 文件，包括 `poll_fn` 自己抛异常的情况——fail-closed，崩溃了就该看起来像崩溃了，不能让 PID 文件继续指向一个已经不存在的循环）。`start_detached()`/`stop()` 是给上层（未来某个"启动 graphlemon 平台"脚本，或人工）用的真实进程管理入口，`start_detached` 用 `subprocess.Popen` 拉起真实独立进程（不是线程，故意如此——两个进程之间不共享失败域是这整个设计的前提），`stop()` 发 `SIGTERM` 并轮询等它自己清理干净，超时不 SIGKILL（留给调用方判断，不擅自做主）。**验证方式**：`RunForegroundTest` 系列在同一个测试进程内直接验证循环/状态写入/`SIGTERM`处理（`SIGTERM` 测试用 `poll_fn` 内部对自己进程 `os.kill(os.getpid(), SIGTERM)` 模拟外部信号，不需要真的起第二个进程）；`RealSubprocessLifecycleTest` 是真实的（不 mock）`python3 -m supervisor.deploy_supervisor` 子进程拉起+`stop()`往返，确认"能独立启停"这条 Done 标准不是靠单元测试逻辑自证的，是真的对着一个真实操作系统进程验证过的。全程用 `GRAPHLEMON_HOME` 环境变量覆盖到临时目录，不碰真实用户的 `~/.graphlemon`。73 个测试（含四个 isolation 模块）全绿。
+  - 覆盖：`supervisor/isolation/` 四模块（真实单测覆盖，不再只是"一次性手工冒烟"）；Deploy Supervisor 进程骨架（`supervisor/deploy_supervisor.py`，尚不含任何真实部署决策逻辑）
 - **U-037 [Risk Gate]** `c_deploy_supervisor_stage`：候选实例拉起（全新目录 checkout，不 `git pull` 原地改）+ 一次性 DB 副本机制（候选实例验证期间绝不接触生产库）
   - 依赖：U-034, U-036
@@ -53,7 +53,7 @@
     {"id": "U-034", "phase": "PH5", "sub_phase": "PH5-S1", "name": "01-M 扩展：deployment/deploy_health_signal + core_engine 枚举（见 development-plan.md 2026-08-02 实现补充）", "classification": "Enabler", "depends_on": ["U-001"], "coverage": {"model": ["deployment", "deploy_health_signal", "bug_record"], "view": [], "controller": [], "process": [], "requirement": []}, "done": "两张新表建表成功（共 17 张）；bug_record 新枚举值向后兼容 U-015/U-018 已有测试；deployment 状态机用 01-M 1.15 完整 15 值枚举；ux_deployment_active 表达式索引强制全局单例非终态约束；bug_record.py 应用层白名单刻意未跟着放开 core_engine，classification+路由逻辑留给 U-035"},
     {"id": "U-035", "phase": "PH5", "sub_phase": "PH5-S1", "name": "self-target worktree 无条件强制 + core_engine 分类审计（见 development-plan.md 2026-08-02 实现补充）", "classification": "Risk Gate", "depends_on": ["U-013", "U-016", "U-034"], "coverage": {"model": [], "view": [], "controller": [], "process": [], "requirement": ["REQ-03"]}, "done": "self-target 时无条件走 worktree（work_graph_orchestrate.run_graph 内强制）；core_engine 分类（src/core_engine_classify.py，路径规则）在 self-target 任务 PASS 收尾时记一条审计用 bug_record；且改动落地后重跑 U-013/REQ-03（worktree 隔离）现有回归用例全部通过。**2026-08-02 用户决策修正**：Deploy Supervisor 触发条件按 supervisor-auto-rollback-design.md §1 字面设计（self-target+PASS 即触发，bug_record 是旁路产物不是前提），而非 03-C/04-P 字面文本要求的\"先有 core_engine bug_record\"——后者会让从不产生 bug_record 的 goal 型自我改进任务永远触发不了 Deploy Supervisor，已用 AskUserQuestion 与用户确认此订正。c_org_loop_route 未改动（fast_lane/full_replan 是 FAIL 路径重试路由，与此处 PASS 路径的蓝绿触发正交），deployment 行创建仍是 U-037 的范围，不在本 Unit 提前做"},
-    {"id": "U-036", "phase": "PH5", "sub_phase": "PH5-S2", "name": "supervisor/isolation 补齐单测 + Deploy Supervisor 进程骨架", "classification": "Enabler", "depends_on": [], "coverage": {"model": [], "view": [], "controller": [], "process": [], "requirement": []}, "done": "四模块各有 test_*.py；Deploy Supervisor 骨架独立进程可启停，不受 Live/Candidate 崩溃影响"},
+    {"id": "U-036", "phase": "PH5", "sub_phase": "PH5-S2", "name": "supervisor/isolation 补齐单测 + Deploy Supervisor 进程骨架（见 development-plan.md 2026-08-02 实现补充）", "classification": "Enabler", "depends_on": [], "coverage": {"model": [], "view": [], "controller": [], "process": [], "requirement": []}, "done": "四模块各有 test_*.py（供 python3 -m unittest discover -s supervisor/isolation 运行，52 测试全绿）；Deploy Supervisor 骨架（supervisor/deploy_supervisor.py）独立进程可启停（真实子进程验证，非 mock），不受 Live/Candidate 崩溃影响；本 Unit 只做进程生命周期骨架，不含任何真实部署决策逻辑（U-037+ 的 poll_fn 挂载点）"},
     {"id": "U-037", "phase": "PH5", "sub_phase": "PH5-S2", "name": "c_deploy_supervisor_stage", "classification": "Risk Gate", "depends_on": ["U-034", "U-036"], "coverage": {"model": ["deployment"], "view": [], "controller": ["c_deploy_supervisor_stage"], "process": [], "requirement": []}, "done": "能生成 deployment 记录并拉起监听临时端口、连一次性 DB 副本的候选进程"},
     {"id": "U-038", "phase": "PH5", "sub_phase": "PH5-S2", "name": "c_deploy_supervisor_health_check", "classification": "Risk Gate", "depends_on": ["U-037"], "coverage": {"model": [], "view": [], "controller": ["c_deploy_supervisor_health_check"], "process": [], "requirement": []}, "done": "process_alive/http_health/canary_task 三项检查全过 -> health_check_passed；任一失败 -> health_check_failed 且 Live 未受影响"},
     {"id": "U-039", "phase": "PH5", "sub_phase": "PH5-S2", "name": "c_deploy_supervisor_cutover", "classification": "Risk Gate", "depends_on": ["U-038"], "coverage": {"model": ["deployment"], "view": [], "controller": ["c_deploy_supervisor_cutover"], "process": [], "requirement": []}, "done": "健康检查通过后完成真实切换；在途 work_node_execution 重启后可恢复；previous_instance_dir 保留至监控窗口结束"},
+"""Deploy Supervisor process skeleton (U-036).
+
+Independent OS process (started via `python3 -m supervisor.deploy_supervisor`),
+pure stdlib, no LLM CLI dependency -- per docs/supervisor-auto-rollback-design.md
+§2's "three independent processes, no shared failure domain" principle:
+Deploy Supervisor crashing must not affect Live/Candidate, and Live/Candidate
+crashing must not affect Deploy Supervisor's own ability to keep observing
+and act.
+
+PID/status file conventions adapted (NOT imported -- see CLAUDE.md's "do not
+import graphloop code directly" rule) from graphloop's hermes_watchdog.py:
+atomic writes (tempfile-in-same-dir + os.replace), liveness via
+`os.kill(pid, 0)`, fail-closed (this skeleton never auto-restarts anything --
+a crashing `poll_fn` stops the loop and marks status='stopped' rather than
+silently dying while still holding the PID file, which would make
+`is_running()` lie).
+
+Scope: this Unit owns process lifecycle only (start/stop/status/PID file).
+The actual "check for staged deployments and act" logic is U-037+'s job,
+injected here as the `poll_fn` callback -- U-036 has no opinion about what
+that logic does, only about running it on a schedule and surviving/reporting
+its own crashes correctly.
+"""
+
+import argparse
+import json
+import os
+import signal
+import subprocess
+import sys
+import time
+
+from src import paths
+
+
+def runtime_dir():
+    d = os.path.join(paths.platform_home(), "deploy_supervisor")
+    os.makedirs(d, exist_ok=True)
+    return d
+
+
+def pid_file_path():
+    return os.path.join(runtime_dir(), "supervisor.pid")
+
+
+def status_file_path():
+    return os.path.join(runtime_dir(), "supervisor.status.json")
+
+
+def _atomic_write_json(path, data):
+    tmp = path + ".{}.tmp".format(os.getpid())
+    with open(tmp, "w", encoding="utf-8") as handle:
+        json.dump(data, handle)
+    os.replace(tmp, path)
+
+
+def is_alive(pid):
+    if not isinstance(pid, int) or pid <= 0:
+        return False
+    try:
+        os.kill(pid, 0)
+    except OSError:
+        return False
+    return True
+
+
+def write_status(status, detail=None):
+    _atomic_write_json(status_file_path(), {
+        "status": status, "pid": os.getpid(), "updated_at": time.time(), "detail": detail,
+    })
+
+
+def read_status():
+    try:
+        with open(status_file_path(), encoding="utf-8") as handle:
+            return json.load(handle)
+    except (OSError, json.JSONDecodeError):
+        return None
+
+
+def is_running():
+    """True if a Deploy Supervisor process is currently alive per the PID
+    file. Doesn't distinguish "never started" from "PID file stale" --
+    callers needing that distinction should also check `read_status()`."""
+    try:
+        with open(pid_file_path(), encoding="utf-8") as handle:
+            pid = int(handle.read().strip())
+    except (OSError, ValueError):
+        return False
+    return is_alive(pid)
+
+
+class _StopRequested(Exception):
+    """Internal: raised by the SIGTERM handler to break run_foreground's loop."""
+
+
+def run_foreground(poll_fn=None, poll_interval=5.0, iterations=None):
+    """Blocking main loop -- calls `poll_fn()` (if given) once per
+    `poll_interval` seconds, up to `iterations` cycles (None = forever, only
+    exits on SIGTERM). Writes a PID file + 'running' status on entry, and
+    unconditionally writes a 'stopped' status + removes the PID file on any
+    exit path (normal completion, SIGTERM, or an unhandled exception from
+    `poll_fn`) -- an exception from `poll_fn` propagates to the caller after
+    cleanup runs, it is not swallowed (fail-closed: a broken poll cycle
+    should surface, not vanish into an infinite silent retry).
+    """
+    def _handle_sigterm(signum, frame):
+        raise _StopRequested()
+
+    previous_handler = signal.signal(signal.SIGTERM, _handle_sigterm)
+    with open(pid_file_path(), "w", encoding="utf-8") as handle:
+        handle.write(str(os.getpid()))
+    write_status("running")
+    try:
+        count = 0
+        while iterations is None or count < iterations:
+            if poll_fn is not None:
+                poll_fn()
+            count += 1
+            if iterations is not None and count >= iterations:
+                break
+            time.sleep(poll_interval)
+    except _StopRequested:
+        pass
+    finally:
+        signal.signal(signal.SIGTERM, previous_handler)
+        write_status("stopped")
+        if os.path.exists(pid_file_path()):
+            os.remove(pid_file_path())
+
+
+def start_detached(poll_interval=5.0):
+    """Launches `python3 -m supervisor.deploy_supervisor` as a real,
+    independent OS process (Popen, not a thread) -- crashing does not take
+    down whatever called this. Returns the child's pid. Raises RuntimeError
+    if a Deploy Supervisor is already running (per its own PID file)."""
+    if is_running():
+        raise RuntimeError("Deploy Supervisor is already running")
+    process = subprocess.Popen(
+        [sys.executable, "-m", "supervisor.deploy_supervisor", "--poll-interval", str(poll_interval)],
+        cwd=paths.repo_root(), stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
+    )
+    return process.pid
+
+
+def stop(timeout=10.0):
+    """Sends SIGTERM to the running Deploy Supervisor (if any) and waits up
+    to `timeout` seconds for it to exit cleanly (PID file removed). Returns
+    True if it stopped, False if it was already not running or didn't stop
+    in time -- fail-closed: this does not SIGKILL on timeout, that's a
+    judgment call left to whoever's driving it, not silently forced here."""
+    try:
+        with open(pid_file_path(), encoding="utf-8") as handle:
+            pid = int(handle.read().strip())
+    except (OSError, ValueError):
+        return False
+    if not is_alive(pid):
+        return False
+    os.kill(pid, signal.SIGTERM)
+    deadline = time.time() + timeout
+    while time.time() < deadline:
+        if not is_running():
+            return True
+        time.sleep(0.1)
+    return False
+
+
+def main():
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--poll-interval", type=float, default=5.0)
+    args = parser.parse_args()
+    run_foreground(poll_interval=args.poll_interval)
+
+
+if __name__ == "__main__":
+    main()
+"""U-036: real unittest coverage for change_manifest.py -- adapted from
+graphloop's test_change_manifest.py (2026-08-02). Module logic is unchanged
+from graphloop ("adapted verbatim", see change_manifest.py's own docstring)
+except graphlemon's `unified_diff` collapses "@@ -l,s +l,s @@" hunk headers
+to a single leading "@@" marker (see that module's own comment) -- the
+"only text gets hunks" assertion below counts "@@" substring occurrences,
+which still holds 1-to-1 with graphlemon's collapsed format for these
+single-hunk fixtures.
+
+Run: python3 -m unittest discover -s supervisor/isolation -p "test_*.py" -v
+"""
+import hashlib
+import os
+import shutil
+import tempfile
+import unittest
+
+from supervisor.isolation import change_manifest as cm
+
+TEXT_A = "hello\n"
+TEXT_A_MODIFIED = "hello modified\n"
+BINARY_BYTES = b"\x89PNG\r\n\x1a\n\x00\x01\x02\x03BINARYDATA\xff\xfe"
+
+
+def sha256_of(data):
+    return hashlib.sha256(data).hexdigest()
+
+
+class ChangeManifestTestCase(unittest.TestCase):
+    def setUp(self):
+        self.source_dir = tempfile.mkdtemp(prefix="cm-src-")
+        self.addCleanup(shutil.rmtree, self.source_dir, ignore_errors=True)
+        self._extra_dirs = []
+        self.addCleanup(self._cleanup_extra_dirs)
+
+    def _cleanup_extra_dirs(self):
+        for path in self._extra_dirs:
+            shutil.rmtree(path, ignore_errors=True)
+
+    def _new_isolated_dir(self):
+        path = tempfile.mkdtemp(prefix="cm-iso-")
+        self._extra_dirs.append(path)
+        return path
+
+    @staticmethod
+    def _write_text(root, rel_path, content):
+        full = os.path.join(root, *rel_path.split("/"))
+        os.makedirs(os.path.dirname(full), exist_ok=True)
+        with open(full, "w", encoding="utf-8") as handle:
+            handle.write(content)
+        return full
+
+    @staticmethod
+    def _write_bytes(root, rel_path, data):
+        full = os.path.join(root, *rel_path.split("/"))
+        os.makedirs(os.path.dirname(full), exist_ok=True)
+        with open(full, "wb") as handle:
+            handle.write(data)
+        return full
+
+    def _copy_source_to_isolated(self):
+        isolated_dir = self._new_isolated_dir()
+        shutil.rmtree(isolated_dir)
+        shutil.copytree(self.source_dir, isolated_dir)
+        return isolated_dir
+
+
+class SnapshotWorkspaceTests(ChangeManifestTestCase):
+    def test_records_text_file_with_hash_size_and_kind(self):
+        self._write_text(self.source_dir, "a.txt", TEXT_A)
+
+        snapshot = cm.snapshot_workspace(self.source_dir)
+
+        entry = snapshot["files"]["a.txt"]
+        self.assertEqual(entry["sha256"], sha256_of(TEXT_A.encode("utf-8")))
+        self.assertEqual(entry["size"], len(TEXT_A.encode("utf-8")))
+        self.assertEqual(entry["kind"], "text")
+
+    def test_records_binary_file_as_binary_kind(self):
+        self._write_bytes(self.source_dir, "logo.bin", BINARY_BYTES)
+
+        snapshot = cm.snapshot_workspace(self.source_dir)
+
+        entry = snapshot["files"]["logo.bin"]
+        self.assertEqual(entry["sha256"], sha256_of(BINARY_BYTES))
+        self.assertEqual(entry["size"], len(BINARY_BYTES))
+        self.assertEqual(entry["kind"], "binary")
+
+    def test_uses_relative_posix_paths_for_nested_files(self):
+        self._write_text(self.source_dir, "subdir/nested/file.txt", "x")
+
+        snapshot = cm.snapshot_workspace(self.source_dir)
+
+        self.assertIn("subdir/nested/file.txt", snapshot["files"])
+
+    def test_ignores_git_pycache_pytest_cache_and_dsstore(self):
+        self._write_text(self.source_dir, "keep.txt", "keep")
+        self._write_text(self.source_dir, ".git/config", "git config")
+        self._write_text(self.source_dir, ".git/objects/ab/cdef", "blob")
+        self._write_text(self.source_dir, "__pycache__/mod.cpython-39.pyc", "pyc")
+        self._write_text(self.source_dir, ".pytest_cache/v/cache/lastfailed", "{}")
+        self._write_text(self.source_dir, ".DS_Store", "junk")
+
+        snapshot = cm.snapshot_workspace(self.source_dir)
+
+        self.assertEqual(set(snapshot["files"]), {"keep.txt"})
+
+    def test_skips_symlinks_but_keeps_their_real_target(self):
+        target = self._write_text(self.source_dir, "real.txt", "real content")
+        link_path = os.path.join(self.source_dir, "link.txt")
+        os.symlink(target, link_path)
+
+        snapshot = cm.snapshot_workspace(self.source_dir)
+
+        self.assertIn("real.txt", snapshot["files"])
+        self.assertNotIn("link.txt", snapshot["files"])
+
+
+class BuildChangeManifestTests(ChangeManifestTestCase):
+    def _baseline_snapshot(self):
+        self._write_text(self.source_dir, "a.txt", TEXT_A)
+        self._write_text(self.source_dir, "b.txt", "world\n")
+        self._write_text(self.source_dir, "c.txt", "keep me\n")
+        return cm.snapshot_workspace(self.source_dir)
+
+    def test_added_modified_deleted_and_unchanged_classification(self):
+        snapshot = self._baseline_snapshot()
+        isolated_dir = self._copy_source_to_isolated()
+
+        self._write_text(isolated_dir, "a.txt", TEXT_A_MODIFIED)
+        os.remove(os.path.join(isolated_dir, "b.txt"))
+        self._write_text(isolated_dir, "d.txt", "brand new\n")
+
+        manifest = cm.build_change_manifest(snapshot, isolated_dir)
+        by_path = {entry["path"]: entry for entry in manifest["changed_files"]}
+
+        self.assertEqual(set(by_path), {"a.txt", "b.txt", "d.txt"})
+        self.assertNotIn("c.txt", by_path)
+
+        modified = by_path["a.txt"]
+        self.assertEqual(modified["status"], "modified")
+        self.assertEqual(modified["old_hash"], sha256_of(TEXT_A.encode("utf-8")))
+        self.assertEqual(modified["new_hash"], sha256_of(TEXT_A_MODIFIED.encode("utf-8")))
+
+        deleted = by_path["b.txt"]
+        self.assertEqual(deleted["status"], "deleted")
+        self.assertIsNone(deleted["new_hash"])
+        self.assertEqual(deleted["old_hash"], sha256_of(b"world\n"))
+
+        added = by_path["d.txt"]
+        self.assertEqual(added["status"], "added")
+        self.assertIsNone(added["old_hash"])
+        self.assertEqual(added["new_hash"], sha256_of(b"brand new\n"))
+
+    def test_unchanged_files_never_appear_in_changed_files(self):
+        snapshot = self._baseline_snapshot()
+        isolated_dir = self._copy_source_to_isolated()
+
+        manifest = cm.build_change_manifest(snapshot, isolated_dir)
+
+        self.assertEqual(manifest["changed_files"], [])
+        self.assertEqual(cm.unified_diff(manifest), "")
+
+    def test_ignored_files_excluded_even_when_added_in_isolated(self):
+        snapshot = self._baseline_snapshot()
+        isolated_dir = self._copy_source_to_isolated()
+
+        self._write_text(isolated_dir, ".git/config", "new git config")
+        self._write_text(isolated_dir, "__pycache__/mod.cpython-39.pyc", "pyc")
+        self._write_text(isolated_dir, ".DS_Store", "junk")
+
+        manifest = cm.build_change_manifest(snapshot, isolated_dir)
+
+        self.assertEqual(manifest["changed_files"], [])
+
+    def test_binary_file_added_is_classified_binary(self):
+        snapshot = self._baseline_snapshot()
+        isolated_dir = self._copy_source_to_isolated()
+        self._write_bytes(isolated_dir, "logo.bin", BINARY_BYTES)
+
+        manifest = cm.build_change_manifest(snapshot, isolated_dir)
+
+        self.assertEqual(len(manifest["changed_files"]), 1)
+        entry = manifest["changed_files"][0]
+        self.assertEqual(entry["path"], "logo.bin")
+        self.assertEqual(entry["status"], "added")
+        self.assertEqual(entry["kind"], "binary")
+        self.assertEqual(entry["new_hash"], sha256_of(BINARY_BYTES))
+
+    def test_changed_files_are_sorted_deterministically(self):
+        snapshot = self._baseline_snapshot()
+        isolated_dir = self._copy_source_to_isolated()
+
+        self._write_text(isolated_dir, "z_file.txt", "z\n")
+        self._write_text(isolated_dir, "0_file.txt", "0\n")
+        self._write_text(isolated_dir, "m_file.txt", "m\n")
+
+        manifest = cm.build_change_manifest(snapshot, isolated_dir)
+        paths = [entry["path"] for entry in manifest["changed_files"]]
+
+        self.assertEqual(paths, sorted(paths))
+        self.assertEqual(paths, ["0_file.txt", "m_file.txt", "z_file.txt"])
+
+    def test_rejects_symlink_found_in_isolated_workspace(self):
+        snapshot = self._baseline_snapshot()
+        isolated_dir = self._copy_source_to_isolated()
+        target = os.path.join(isolated_dir, "a.txt")
+        os.symlink(target, os.path.join(isolated_dir, "sneaky_link.txt"))
+
+        with self.assertRaises(cm.ChangeManifestError):
+            cm.build_change_manifest(snapshot, isolated_dir)
+
+    def test_rejects_symlink_escaping_isolated_workspace(self):
+        snapshot = self._baseline_snapshot()
+        isolated_dir = self._copy_source_to_isolated()
+        outside_dir = self._new_isolated_dir()
+        outside_file = self._write_text(outside_dir, "outside.txt", "outside")
+        os.symlink(outside_file, os.path.join(isolated_dir, "escape_link.txt"))
+
+        with self.assertRaises(cm.ChangeManifestError):
+            cm.build_change_manifest(snapshot, isolated_dir)
+
+    def test_rejects_tampered_snapshot_with_traversal_path(self):
+        snapshot = self._baseline_snapshot()
+        isolated_dir = self._copy_source_to_isolated()
+
+        for bad_path in ("../evil.txt", "/etc/passwd", "sub/../../escape.txt"):
+            with self.subTest(bad_path=bad_path):
+                tampered = {"root": snapshot["root"], "files": dict(snapshot["files"])}
+                tampered["files"][bad_path] = {"sha256": "0" * 64, "size": 0, "kind": "text"}
+                with self.assertRaises(cm.ChangeManifestError):
+                    cm.build_change_manifest(tampered, isolated_dir)
+
+
+class UnifiedDiffTests(ChangeManifestTestCase):
+    def test_modified_text_file_produces_readable_unified_diff(self):
+        self._write_text(self.source_dir, "a.txt", TEXT_A)
+        snapshot = cm.snapshot_workspace(self.source_dir)
+        isolated_dir = self._copy_source_to_isolated()
+        self._write_text(isolated_dir, "a.txt", TEXT_A_MODIFIED)
+
+        manifest = cm.build_change_manifest(snapshot, isolated_dir)
+        diff_text = cm.unified_diff(manifest)
+        lines = diff_text.splitlines()
+
+        self.assertTrue(any(l.startswith("---") and "a.txt" in l for l in lines))
+        self.assertTrue(any(l.startswith("+++") and "a.txt" in l for l in lines))
+        self.assertTrue(any(l.startswith("@@") for l in lines))
+        self.assertTrue(any(l.startswith("-") and "hello" in l and not l.startswith("---") for l in lines))
+        self.assertTrue(any(l.startswith("+") and "hello modified" in l and not l.startswith("+++") for l in lines))
+
+    def test_added_text_file_diffs_against_dev_null(self):
+        snapshot = cm.snapshot_workspace(self.source_dir)
+        isolated_dir = self._copy_source_to_isolated()
+        self._write_text(isolated_dir, "new.txt", "brand new content\n")
+
+        manifest = cm.build_change_manifest(snapshot, isolated_dir)
+        diff_text = cm.unified_diff(manifest)
+        lines = diff_text.splitlines()
+
+        self.assertTrue(any("/dev/null" in l for l in lines))
+        self.assertTrue(any(l.startswith("+") and "brand new content" in l for l in lines))
+        removed_content_lines = [l for l in lines if l.startswith("-") and not l.startswith("---")]
+        self.assertEqual(removed_content_lines, [])
+
+    def test_deleted_text_file_diffs_against_dev_null(self):
+        self._write_text(self.source_dir, "gone.txt", "will be deleted\n")
+        snapshot = cm.snapshot_workspace(self.source_dir)
+        isolated_dir = self._copy_source_to_isolated()
+        os.remove(os.path.join(isolated_dir, "gone.txt"))
+
+        manifest = cm.build_change_manifest(snapshot, isolated_dir)
+        diff_text = cm.unified_diff(manifest)
+        lines = diff_text.splitlines()
+
+        self.assertTrue(any("/dev/null" in l for l in lines))
+        self.assertTrue(any(l.startswith("-") and "will be deleted" in l for l in lines))
+        added_content_lines = [l for l in lines if l.startswith("+") and not l.startswith("+++")]
+        self.assertEqual(added_content_lines, [])
+
+    def test_binary_file_diff_is_marker_not_content_diff(self):
+        snapshot = cm.snapshot_workspace(self.source_dir)
+        isolated_dir = self._copy_source_to_isolated()
+        self._write_bytes(isolated_dir, "logo.bin", BINARY_BYTES)
+
+        manifest = cm.build_change_manifest(snapshot, isolated_dir)
+        diff_text = cm.unified_diff(manifest)
+
+        self.assertIn("Binary files", diff_text)
+        self.assertIn("logo.bin", diff_text)
+        self.assertNotIn("@@", diff_text)
+
+    def test_text_and_binary_changes_together_only_text_gets_hunks(self):
+        self._write_text(self.source_dir, "a.txt", TEXT_A)
+        self._write_text(self.source_dir, "gone.txt", "will be deleted\n")
+        snapshot = cm.snapshot_workspace(self.source_dir)
+        isolated_dir = self._copy_source_to_isolated()
+
+        self._write_text(isolated_dir, "a.txt", TEXT_A_MODIFIED)
+        os.remove(os.path.join(isolated_dir, "gone.txt"))
+        self._write_text(isolated_dir, "new.txt", "added text\n")
+        self._write_bytes(isolated_dir, "logo.bin", BINARY_BYTES)
+
+        manifest = cm.build_change_manifest(snapshot, isolated_dir)
+        diff_text = cm.unified_diff(manifest)
+
+        text_change_count = sum(1 for entry in manifest["changed_files"] if entry["kind"] == "text")
+        self.assertEqual(diff_text.count("@@"), text_change_count)
+        self.assertIn("Binary files", diff_text)
+
+
+if __name__ == "__main__":
+    unittest.main()
+"""U-036: real unittest coverage for isolated_workspace.py -- adapted from
+graphloop's test_isolated_coding_workspace.py (2026-08-02). Module logic is
+unchanged from graphloop (see isolated_workspace.py's own docstring); only
+the import path changed to match this package's location.
+
+Run: python3 -m unittest discover -s supervisor/isolation -p "test_*.py" -v
+"""
+import os
+import shutil
+import tempfile
+import unittest
+
+from supervisor.isolation import isolated_workspace as iso
+
+
+class IsolateWorkspaceTests(unittest.TestCase):
+    def setUp(self):
+        self.source_dir = tempfile.mkdtemp(prefix="iso-src-")
+        self.addCleanup(shutil.rmtree, self.source_dir, ignore_errors=True)
+
+        self.source_file = os.path.join(self.source_dir, "hello.txt")
+        with open(self.source_file, "w", encoding="utf-8") as handle:
+            handle.write("hello world")
+
+        self._copies_to_clean = []
+        self.addCleanup(self._cleanup_copies)
+
+    def _cleanup_copies(self):
+        for path in self._copies_to_clean:
+            shutil.rmtree(path, ignore_errors=True)
+
+    def test_returns_separate_path_leaves_source_unchanged_and_copies_file(self):
+        copied_dir = iso.isolate_workspace(self.source_dir)
+        self._copies_to_clean.append(copied_dir)
+
+        self.assertNotEqual(os.path.abspath(copied_dir), os.path.abspath(self.source_dir))
+
+        self.assertTrue(os.path.isdir(self.source_dir))
+        with open(self.source_file, "r", encoding="utf-8") as handle:
+            self.assertEqual(handle.read(), "hello world")
+
+        copied_file = os.path.join(copied_dir, "hello.txt")
+        self.assertTrue(os.path.isfile(copied_file))
+        with open(copied_file, "r", encoding="utf-8") as handle:
+            self.assertEqual(handle.read(), "hello world")
+
+    def test_parent_dir_overrides_os_default_temp_location(self):
+        custom_parent = tempfile.mkdtemp(prefix="iso-parent-")
+        self.addCleanup(shutil.rmtree, custom_parent, ignore_errors=True)
+
+        copied_dir = iso.isolate_workspace(self.source_dir, parent_dir=custom_parent)
+        self._copies_to_clean.append(copied_dir)
+
+        self.assertEqual(os.path.dirname(copied_dir), custom_parent)
+
+    def test_parent_dir_created_if_missing(self):
+        base = tempfile.mkdtemp(prefix="iso-parent-")
+        self.addCleanup(shutil.rmtree, base, ignore_errors=True)
+        custom_parent = os.path.join(base, "nested", "dir")
+
+        copied_dir = iso.isolate_workspace(self.source_dir, parent_dir=custom_parent)
+        self._copies_to_clean.append(copied_dir)
+
+        self.assertTrue(os.path.isdir(custom_parent))
+        self.assertEqual(os.path.dirname(copied_dir), custom_parent)
+
+    def test_rejects_nonexistent_source_dir(self):
+        with self.assertRaises(iso.IsolatedWorkspaceError):
+            iso.isolate_workspace(os.path.join(self.source_dir, "does-not-exist"))
+
+    def test_rejects_empty_source_dir_argument(self):
+        with self.assertRaises(iso.IsolatedWorkspaceError):
+            iso.isolate_workspace("")
+
+
+if __name__ == "__main__":
+    unittest.main()
+"""U-036: real unittest coverage for selective_apply.py -- adapted from
+graphloop's test_selective_apply.py (2026-08-02). Module logic is unchanged
+from graphloop except the internal `import change_manifest as cm` ->
+relative package import (see selective_apply.py's own docstring); the
+result key is `manifest_id` (not `digest`) in this port.
+
+Run: python3 -m unittest discover -s supervisor/isolation -p "test_*.py" -v
+"""
+import json
+import os
+import shutil
+import tempfile
+import unittest
+
+from supervisor.isolation import change_manifest as cm
+from supervisor.isolation import selective_apply as sa
+
+
+class SelectiveApplyTestCase(unittest.TestCase):
+    @staticmethod
+    def _write(root, rel_path, data):
+        full = os.path.join(root, *rel_path.split("/"))
+        os.makedirs(os.path.dirname(full), exist_ok=True)
+        with open(full, "wb") as handle:
+            handle.write(data)
+        return full
+
+    def _build_scenario(self):
+        source_dir = tempfile.mkdtemp(prefix="sa-src-")
+        isolated_dir = tempfile.mkdtemp(prefix="sa-iso-")
+        self.addCleanup(shutil.rmtree, source_dir, ignore_errors=True)
+        self.addCleanup(shutil.rmtree, isolated_dir, ignore_errors=True)
+
+        self._write(source_dir, "keep.txt", b"keep\n")
+        self._write(isolated_dir, "keep.txt", b"keep\n")
+
+        self._write(source_dir, "mod.txt", b"original\n")
+        self._write(isolated_dir, "mod.txt", b"changed\n")
+
+        self._write(source_dir, "old.txt", b"old content\n")
+
+        self._write(isolated_dir, "new.txt", b"new content\n")
+
+        source_snapshot = cm.snapshot_workspace(source_dir)
+        manifest = cm.build_change_manifest(source_snapshot, isolated_dir)
+        return source_dir, isolated_dir, manifest, source_snapshot
+
+    def test_approved_paths_none_raises(self):
+        source_dir, isolated_dir, manifest, before = self._build_scenario()
+        with self.assertRaises(sa.SelectiveApplyError):
+            sa.apply_manifest(manifest, source_dir, isolated_dir, None)
+        after = cm.snapshot_workspace(source_dir)
+        self.assertEqual(before, after)
+
+    def test_rejects_path_not_in_manifest_and_leaves_source_untouched(self):
+        source_dir, isolated_dir, manifest, before = self._build_scenario()
+        with self.assertRaises(sa.SelectiveApplyError):
+            sa.apply_manifest(manifest, source_dir, isolated_dir, ["mod.txt", "not_in_manifest.txt"])
+        after = cm.snapshot_workspace(source_dir)
+        self.assertEqual(before, after)
+
+    def test_rejects_absolute_and_traversal_paths(self):
+        source_dir, isolated_dir, manifest, before = self._build_scenario()
+        for bad_path in ["/etc/passwd", "../escape.txt", "sub/../../escape.txt"]:
+            with self.assertRaises(sa.SelectiveApplyError):
+                sa.apply_manifest(manifest, source_dir, isolated_dir, [bad_path])
+        after = cm.snapshot_workspace(source_dir)
+        self.assertEqual(before, after)
+
+    def test_applies_only_approved_subset_and_preserves_the_rest(self):
+        source_dir, isolated_dir, manifest, _ = self._build_scenario()
+        result = sa.apply_manifest(manifest, source_dir, isolated_dir, ["mod.txt"])
+
+        self.assertEqual(result["applied"], ["mod.txt"])
+        self.assertEqual(sorted(result["skipped"]), ["new.txt", "old.txt"])
+        self.assertEqual(result["already_applied"], [])
+
+        with open(os.path.join(source_dir, "mod.txt"), "rb") as handle:
+            self.assertEqual(handle.read(), b"changed\n")
+        self.assertFalse(os.path.exists(os.path.join(source_dir, "new.txt")))
+        with open(os.path.join(source_dir, "old.txt"), "rb") as handle:
+            self.assertEqual(handle.read(), b"old content\n")
+        with open(os.path.join(source_dir, "keep.txt"), "rb") as handle:
+            self.assertEqual(handle.read(), b"keep\n")
+
+    def test_apply_add_modify_delete_full_manifest(self):
+        source_dir, isolated_dir, manifest, _ = self._build_scenario()
+        result = sa.apply_manifest(manifest, source_dir, isolated_dir, ["mod.txt", "new.txt", "old.txt"])
+
+        self.assertEqual(sorted(result["applied"]), ["mod.txt", "new.txt", "old.txt"])
+        self.assertEqual(result["skipped"], [])
+        self.assertEqual(result["already_applied"], [])
+
+        with open(os.path.join(source_dir, "mod.txt"), "rb") as handle:
+            self.assertEqual(handle.read(), b"changed\n")
+        with open(os.path.join(source_dir, "new.txt"), "rb") as handle:
+            self.assertEqual(handle.read(), b"new content\n")
+        self.assertFalse(os.path.exists(os.path.join(source_dir, "old.txt")))
+        with open(os.path.join(source_dir, "keep.txt"), "rb") as handle:
+            self.assertEqual(handle.read(), b"keep\n")
+
+    def test_source_drift_rejected_before_any_write(self):
+        source_dir, isolated_dir, manifest, expected_snapshot = self._build_scenario()
+
+        self._write(source_dir, "keep.txt", b"drifted!\n")
+
+        with self.assertRaises(sa.SelectiveApplyError):
+            sa.apply_manifest(
+                manifest, source_dir, isolated_dir, ["mod.txt", "new.txt", "old.txt"],
+                expected_source_snapshot=expected_snapshot,
+            )
+
+        with open(os.path.join(source_dir, "mod.txt"), "rb") as handle:
+            self.assertEqual(handle.read(), b"original\n")
+        self.assertFalse(os.path.exists(os.path.join(source_dir, "new.txt")))
+        with open(os.path.join(source_dir, "old.txt"), "rb") as handle:
+            self.assertEqual(handle.read(), b"old content\n")
+        with open(os.path.join(source_dir, "keep.txt"), "rb") as handle:
+            self.assertEqual(handle.read(), b"drifted!\n")
+
+    def test_idempotent_second_apply_reports_already_applied(self):
+        source_dir, isolated_dir, manifest, _ = self._build_scenario()
+        approved = ["mod.txt", "new.txt", "old.txt"]
+
+        first = sa.apply_manifest(manifest, source_dir, isolated_dir, approved)
+        self.assertEqual(sorted(first["applied"]), sorted(approved))
+
+        snapshot_after_first = cm.snapshot_workspace(source_dir)
+
+        second = sa.apply_manifest(manifest, source_dir, isolated_dir, approved)
+        self.assertEqual(second["applied"], [])
+        self.assertEqual(sorted(second["already_applied"]), sorted(approved))
+
+        snapshot_after_second = cm.snapshot_workspace(source_dir)
+        self.assertEqual(snapshot_after_first, snapshot_after_second)
+
+    def test_symlink_destination_rejected(self):
+        source_dir = tempfile.mkdtemp(prefix="sa-src-")
+        isolated_dir = tempfile.mkdtemp(prefix="sa-iso-")
+        self.addCleanup(shutil.rmtree, source_dir, ignore_errors=True)
+        self.addCleanup(shutil.rmtree, isolated_dir, ignore_errors=True)
+
+        real_target = self._write(source_dir, "real.txt", b"real\n")
+        symlink_path = os.path.join(source_dir, "linked.txt")
+        os.symlink(real_target, symlink_path)
+
+        self._write(isolated_dir, "linked.txt", b"attacker controlled\n")
+
+        source_snapshot = cm.snapshot_workspace(source_dir)
+        manifest = cm.build_change_manifest(source_snapshot, isolated_dir)
+        self.assertIn("linked.txt", [entry["path"] for entry in manifest["changed_files"]])
+
+        with self.assertRaises(sa.SelectiveApplyError):
+            sa.apply_manifest(manifest, source_dir, isolated_dir, ["linked.txt"])
+
+        self.assertTrue(os.path.islink(symlink_path))
+        self.assertEqual(os.readlink(symlink_path), real_target)
+        with open(real_target, "rb") as handle:
+            self.assertEqual(handle.read(), b"real\n")
+
+    def test_result_is_json_serializable_and_carries_stable_identifier(self):
+        source_dir, isolated_dir, manifest, _ = self._build_scenario()
+        result = sa.apply_manifest(manifest, source_dir, isolated_dir, ["mod.txt"])
+
+        json.dumps(result)
+
+        self.assertIn("manifest_id", result)
+        self.assertIsInstance(result["manifest_id"], str)
+        self.assertTrue(result["manifest_id"])
+
+        second = sa.apply_manifest(manifest, source_dir, isolated_dir, ["mod.txt"])
+        self.assertEqual(result["manifest_id"], second["manifest_id"])
+
+
+if __name__ == "__main__":
+    unittest.main()
+"""U-036: real unittest coverage for workspace_security.py -- adapted from
+graphloop's test_workspace_security.py (2026-08-02). Module logic is
+unchanged from graphloop ("adapted verbatim", see workspace_security.py's
+own docstring); only the import path changed.
+
+Run: python3 -m unittest discover -s supervisor/isolation -p "test_*.py" -v
+"""
+
+import os
+import shutil
+import sys
+import tempfile
+import unittest
+
+from supervisor.isolation import workspace_security as ws
+
+
+class ValidateArtifactsTests(unittest.TestCase):
+    def setUp(self):
+        self.workspace = tempfile.mkdtemp(prefix="wssec-workspace-")
+        self.addCleanup(shutil.rmtree, self.workspace, ignore_errors=True)
+
+        self.outside_dir = tempfile.mkdtemp(prefix="wssec-outside-")
+        self.addCleanup(shutil.rmtree, self.outside_dir, ignore_errors=True)
+
+    def _write(self, relative_path, content="content"):
+        full_path = os.path.join(self.workspace, relative_path)
+        os.makedirs(os.path.dirname(full_path), exist_ok=True)
+        with open(full_path, "w", encoding="utf-8") as handle:
+            handle.write(content)
+        return full_path
+
+    def test_accepts_and_normalizes_plain_relative_artifacts(self):
+        self._write("output.txt")
+        self._write(os.path.join("nested", "dir", "file.txt"))
+
+        result = ws.validate_artifacts(["output.txt", "nested/dir/file.txt"], self.workspace)
+
+        self.assertEqual(result, ["output.txt", "nested/dir/file.txt"])
+
+    def test_normalizes_redundant_leading_dot_slash(self):
+        self._write("output.txt")
+
+        result = ws.validate_artifacts(["./output.txt"], self.workspace)
+
+        self.assertEqual(result, ["output.txt"])
+
+    def test_rejects_non_string_item(self):
+        self._write("output.txt")
+
+        with self.assertRaises(ws.WorkspaceSecurityError):
+            ws.validate_artifacts(["output.txt", 42], self.workspace)
+
+    def test_rejects_non_list_artifacts_argument(self):
+        with self.assertRaises(ws.WorkspaceSecurityError):
+            ws.validate_artifacts("output.txt", self.workspace)
+
+    def test_rejects_absolute_path_artifact(self):
+        absolute = self._write("output.txt")
+
+        with self.assertRaises(ws.WorkspaceSecurityError):
+            ws.validate_artifacts([absolute], self.workspace)
+
+    def test_rejects_dot_dot_traversal_even_when_target_exists(self):
+        self._write("output.txt")
+        os.makedirs(os.path.join(self.workspace, "sibling"), exist_ok=True)
+
+        with self.assertRaises(ws.WorkspaceSecurityError):
+            ws.validate_artifacts(["sibling/../output.txt"], self.workspace)
+
+    def test_rejects_dot_dot_traversal_escaping_workspace(self):
+        secret = os.path.join(self.outside_dir, "secret.txt")
+        with open(secret, "w", encoding="utf-8") as handle:
+            handle.write("secret")
+
+        with self.assertRaises(ws.WorkspaceSecurityError):
+            ws.validate_artifacts(
+                [os.path.join("..", os.path.basename(self.outside_dir), "secret.txt")],
+                self.workspace,
+            )
+
+    def test_rejects_missing_artifact(self):
+        with self.assertRaises(ws.WorkspaceSecurityError):
+            ws.validate_artifacts(["does_not_exist.txt"], self.workspace)
+
+    def test_rejects_symlink_artifact(self):
+        real_file = self._write("real.txt")
+        symlink_path = os.path.join(self.workspace, "link.txt")
+        os.symlink(real_file, symlink_path)
+        self.addCleanup(lambda: os.path.exists(symlink_path) and os.remove(symlink_path))
+
+        with self.assertRaises(ws.WorkspaceSecurityError):
+            ws.validate_artifacts(["link.txt"], self.workspace)
+
+    def test_rejects_artifact_resolving_outside_workspace_via_symlinked_directory(self):
+        secret = os.path.join(self.outside_dir, "secret.txt")
+        with open(secret, "w", encoding="utf-8") as handle:
+            handle.write("secret")
+
+        symlinked_dir = os.path.join(self.workspace, "escape")
+        os.symlink(self.outside_dir, symlinked_dir, target_is_directory=True)
+        self.addCleanup(lambda: os.path.exists(symlinked_dir) and os.remove(symlinked_dir))
+
+        with self.assertRaises(ws.WorkspaceSecurityError):
+            ws.validate_artifacts(["escape/secret.txt"], self.workspace)
+
+    def test_rejects_empty_string_artifact(self):
+        with self.assertRaises(ws.WorkspaceSecurityError):
+            ws.validate_artifacts([""], self.workspace)
+
+
+class ValidateSubprocessCommandTests(unittest.TestCase):
+    def setUp(self):
+        self.workspace = tempfile.mkdtemp(prefix="wssec-cmd-workspace-")
+        self.addCleanup(shutil.rmtree, self.workspace, ignore_errors=True)
+
+        self.adapter_script = os.path.join(self.workspace, "existing_adapter.py")
+        with open(self.adapter_script, "w", encoding="utf-8") as handle:
+            handle.write("print('ok')\n")
+
+    def test_accepts_python_executable_with_existing_script(self):
+        result = ws.validate_subprocess_command([sys.executable, self.adapter_script])
+
+        self.assertEqual(result, [sys.executable, self.adapter_script])
+
+    def test_rejects_empty_command(self):
+        with self.assertRaises(ws.WorkspaceSecurityError):
+            ws.validate_subprocess_command([])
+
+    def test_rejects_non_list_command(self):
+        with self.assertRaises(ws.WorkspaceSecurityError):
+            ws.validate_subprocess_command("{} {}".format(sys.executable, self.adapter_script))
+
+    def test_rejects_non_string_argument(self):
+        with self.assertRaises(ws.WorkspaceSecurityError):
+            ws.validate_subprocess_command([sys.executable, self.adapter_script, 7])
+
+    def test_rejects_shell_metacharacters_in_arguments(self):
+        dangerous_args = [
+            "foo; rm -rf /", "foo && rm -rf /", "foo | rm -rf /",
+            "foo > /etc/passwd", "foo `rm -rf /`", "foo $(rm -rf /)",
+        ]
+        for dangerous_arg in dangerous_args:
+            with self.subTest(dangerous_arg=dangerous_arg):
+                with self.assertRaises(ws.WorkspaceSecurityError):
+                    ws.validate_subprocess_command([sys.executable, self.adapter_script, dangerous_arg])
+
+    def test_rejects_bare_shell_invocation(self):
+        with self.assertRaises(ws.WorkspaceSecurityError):
+            ws.validate_subprocess_command(["/bin/sh", "-c", "echo hi"])
+
+    def test_rejects_script_argument_with_disallowed_extension(self):
+        non_script = os.path.join(self.workspace, "not_a_script.txt")
+        with open(non_script, "w", encoding="utf-8") as handle:
+            handle.write("not python")
+
+        with self.assertRaises(ws.WorkspaceSecurityError):
+            ws.validate_subprocess_command([sys.executable, non_script])
+
+    def test_rejects_script_argument_that_does_not_exist(self):
+        missing_script = os.path.join(self.workspace, "missing_adapter.py")
+
+        with self.assertRaises(ws.WorkspaceSecurityError):
+            ws.validate_subprocess_command([sys.executable, missing_script])
+
+    def test_rejects_newline_in_argument(self):
+        with self.assertRaises(ws.WorkspaceSecurityError):
+            ws.validate_subprocess_command([sys.executable, self.adapter_script, "foo\nbar"])
+
+
+if __name__ == "__main__":
+    unittest.main()
+"""U-036 tests: Deploy Supervisor process skeleton -- lifecycle only (PID
+file, status file, start/stop), no actual deploy logic (that's U-037+).
+GRAPHLEMON_HOME is always overridden to a throwaway tempdir so these tests
+never touch the real user's ~/.graphlemon.
+
+Run: python3 -m unittest discover -s supervisor -p "test_*.py" -v
+"""
+
+import os
+import shutil
+import signal
+import subprocess
+import sys
+import tempfile
+import time
+import unittest
+
+from supervisor import deploy_supervisor as ds
+from src import paths
+
+
+class DeploySupervisorTestBase(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.mkdtemp(prefix="deploy-supervisor-test-")
+        self.addCleanup(shutil.rmtree, self.tmp, ignore_errors=True)
+        self.old_home = os.environ.get("GRAPHLEMON_HOME")
+        os.environ["GRAPHLEMON_HOME"] = self.tmp
+
+    def tearDown(self):
+        if self.old_home is None:
+            os.environ.pop("GRAPHLEMON_HOME", None)
+        else:
+            os.environ["GRAPHLEMON_HOME"] = self.old_home
+
+
+class RuntimePathsTest(DeploySupervisorTestBase):
+    def test_runtime_dir_created_under_platform_home(self):
+        d = ds.runtime_dir()
+        self.assertTrue(os.path.isdir(d))
+        self.assertTrue(d.startswith(paths.platform_home()))
+
+    def test_pid_and_status_file_paths_live_under_runtime_dir(self):
+        self.assertEqual(os.path.dirname(ds.pid_file_path()), ds.runtime_dir())
+        self.assertEqual(os.path.dirname(ds.status_file_path()), ds.runtime_dir())
+
+
+class IsAliveTest(unittest.TestCase):
+    def test_current_process_is_alive(self):
+        self.assertTrue(ds.is_alive(os.getpid()))
+
+    def test_zero_or_negative_pid_is_not_alive(self):
+        self.assertFalse(ds.is_alive(0))
+        self.assertFalse(ds.is_alive(-1))
+
+    def test_non_int_is_not_alive(self):
+        self.assertFalse(ds.is_alive("not-a-pid"))
+
+    def test_a_pid_that_does_not_exist_is_not_alive(self):
+        # PID_MAX on Linux/macOS is well below this; astronomically unlikely
+        # to collide with a real running process.
+        self.assertFalse(ds.is_alive(2**30))
+
+
+class StatusFileTest(DeploySupervisorTestBase):
+    def test_write_then_read_status_round_trips(self):
+        ds.write_status("running", detail="test detail")
+        status = ds.read_status()
+        self.assertEqual(status["status"], "running")
+        self.assertEqual(status["detail"], "test detail")
+        self.assertEqual(status["pid"], os.getpid())
+
+    def test_read_status_returns_none_when_no_file_yet(self):
+        self.assertIsNone(ds.read_status())
+
+    def test_write_status_is_atomic_no_tmp_file_left_behind(self):
+        ds.write_status("running")
+        entries = os.listdir(ds.runtime_dir())
+        self.assertEqual(entries, ["supervisor.status.json"])
+
+
+class IsRunningTest(DeploySupervisorTestBase):
+    def test_false_when_no_pid_file(self):
+        self.assertFalse(ds.is_running())
+
+    def test_true_when_pid_file_points_to_a_live_process(self):
+        with open(ds.pid_file_path(), "w", encoding="utf-8") as handle:
+            handle.write(str(os.getpid()))
+        self.assertTrue(ds.is_running())
+
+    def test_false_when_pid_file_points_to_a_dead_process(self):
+        with open(ds.pid_file_path(), "w", encoding="utf-8") as handle:
+            handle.write(str(2**30))
+        self.assertFalse(ds.is_running())
+
+    def test_false_when_pid_file_is_corrupt(self):
+        with open(ds.pid_file_path(), "w", encoding="utf-8") as handle:
+            handle.write("not-a-number")
+        self.assertFalse(ds.is_running())
+
+
+class RunForegroundTest(DeploySupervisorTestBase):
+    def test_writes_pid_file_and_running_status_during_the_loop(self):
+        seen_running = []
+
+        def poll_fn():
+            seen_running.append(ds.is_running())
+            seen_running.append(ds.read_status()["status"])
+
+        ds.run_foreground(poll_fn=poll_fn, poll_interval=0, iterations=1)
+
+        self.assertEqual(seen_running, [True, "running"])
+
+    def test_calls_poll_fn_exactly_iterations_times(self):
+        calls = []
+        ds.run_foreground(poll_fn=lambda: calls.append(1), poll_interval=0, iterations=3)
+        self.assertEqual(len(calls), 3)
+
+    def test_writes_stopped_status_and_removes_pid_file_on_normal_completion(self):
+        ds.run_foreground(poll_fn=None, poll_interval=0, iterations=1)
+        self.assertEqual(ds.read_status()["status"], "stopped")
+        self.assertFalse(os.path.exists(ds.pid_file_path()))
+
+    def test_sigterm_breaks_the_loop_and_still_writes_stopped_status(self):
+        calls = []
+
+        def poll_fn():
+            calls.append(1)
+            if len(calls) == 2:
+                os.kill(os.getpid(), signal.SIGTERM)
+
+        # iterations=None (would run forever) -- SIGTERM is the only exit.
+        ds.run_foreground(poll_fn=poll_fn, poll_interval=0, iterations=None)
+
+        self.assertEqual(len(calls), 2)
+        self.assertEqual(ds.read_status()["status"], "stopped")
+        self.assertFalse(os.path.exists(ds.pid_file_path()))
+
+    def test_poll_fn_exception_propagates_but_cleanup_still_runs(self):
+        def poll_fn():
+            raise RuntimeError("boom")
+
+        with self.assertRaises(RuntimeError):
+            ds.run_foreground(poll_fn=poll_fn, poll_interval=0, iterations=1)
+
+        self.assertEqual(ds.read_status()["status"], "stopped")
+        self.assertFalse(os.path.exists(ds.pid_file_path()))
+
+
+class RealSubprocessLifecycleTest(DeploySupervisorTestBase):
+    """Not mocked -- actually spawns `python3 -m supervisor.deploy_supervisor`
+    as a real independent OS process and stops it for real, per this Unit's
+    Done criterion ("骨架能独立启停")."""
+
+    def test_start_detached_then_stop(self):
+        env = dict(os.environ)
+        # start_detached itself calls is_running()/Popen using the current
+        # process's env var (already patched by setUp); the child process
+        # inherits it via Popen's default env=None -> os.environ passthrough.
+        pid = ds.start_detached(poll_interval=0.1)
+        try:
+            deadline = time.time() + 5
+            while time.time() < deadline and not ds.is_running():
+                time.sleep(0.05)
+            self.assertTrue(ds.is_running())
+            self.assertTrue(ds.is_alive(pid))
+
+            stopped = ds.stop(timeout=5)
+            self.assertTrue(stopped)
+            self.assertFalse(ds.is_running())
+        finally:
+            if ds.is_alive(pid):
+                os.kill(pid, signal.SIGKILL)
+
+    def test_start_detached_twice_raises_while_already_running(self):
+        pid = ds.start_detached(poll_interval=0.1)
+        try:
+            deadline = time.time() + 5
+            while time.time() < deadline and not ds.is_running():
+                time.sleep(0.05)
+            self.assertTrue(ds.is_running())
+            with self.assertRaises(RuntimeError):
+                ds.start_detached(poll_interval=0.1)
+        finally:
+            ds.stop(timeout=5)
+            if ds.is_alive(pid):
+                os.kill(pid, signal.SIGKILL)
+
+    def test_stop_returns_false_when_nothing_is_running(self):
+        self.assertFalse(ds.stop(timeout=1))
+
+
+if __name__ == "__main__":
+    unittest.main()