diff --git a/Agent.md b/Agent.md
index c5a0508..419ccfc 100644
--- a/Agent.md
+++ b/Agent.md
@@ -112,7 +112,7 @@ Community needs voiced in HN agent-UI discussions map directly to EMRG's design:
 pkill -f "emrg.server"; rm -f ~/.emrg/emrgd.port; python -m emrg
 ```
 
-Python: `uv run pytest tests/ -v` (736) — import check: `uv run python -c "from emrg.client.app import run_client"`
+Python: `uv run pytest tests/ -v` (744) — import check: `uv run python -c "from emrg.client.app import run_client"`
 GUI: `cd emrg/gui && npm test` (225: 43 daemon_client + 19 conn-manager + 22 app-commands + 104 renderer smoke + 15 i18n + 7 integration + 3 commands + 5 build-config + 7 gui-state) — syntax: `node --check main.js preload.js daemon_client.js renderer/js/*.js`
 CI: `uv run pytest` + GUI tests + **actionlint workflow lint** (`rhysd/actionlint@v1.7.12` gate, #444 — workflow 解析错误在 PR CI 即失败，如 `if:` secrets 上下文)
 Re-trigger: `scripts/re-trigger-ci.sh [branch]` (workflow_dispatch, #527 — 替代空 commit 重触发：Actions outage 会整段丢弃 push 事件，dispatch 走 API 路径不受影响)
diff --git a/emrg/server/daemon.py b/emrg/server/daemon.py
index d595a1e..f306533 100644
--- a/emrg/server/daemon.py
+++ b/emrg/server/daemon.py
@@ -1217,6 +1217,83 @@ async def _process_message(
                     "error": "scheduler not running",
                 })
 
+        elif msg_type == "task_create":
+            # rant 2026-08-12T18:23:15 P2 — task CRUD with hot reload
+            if not self._scheduler:
+                await self._send(ws, {"type": "task_result", "error": "scheduler not running"})
+                return
+            ok, res = self._scheduler.task_create(
+                name=msg.get("name", "").strip(),
+                task_type=msg.get("type", "").strip(),
+                project=msg.get("project", "").strip(),
+                interval=msg.get("interval"),
+                enabled=msg.get("enabled", True),
+                repo=msg.get("repo"),
+                description=msg.get("description"),
+            )
+            if not ok:
+                await self._send(ws, {"type": "task_result", "error": res})
+                return
+            summary = await self._scheduler.apply_tasks(self._scheduler._load_tasks())
+            await self._send(ws, {"type": "task_result", "ok": True, "task": res, "summary": summary})
+
+        elif msg_type == "task_update":
+            if not self._scheduler:
+                await self._send(ws, {"type": "task_result", "error": "scheduler not running"})
+                return
+            fields = {k: msg[k] for k in ("type", "project", "interval", "enabled", "repo", "description") if k in msg}
+            ok, res = self._scheduler.task_update(msg.get("name", "").strip(), **fields)
+            if not ok:
+                await self._send(ws, {"type": "task_result", "error": res})
+                return
+            summary = await self._scheduler.apply_tasks(self._scheduler._load_tasks())
+            await self._send(ws, {"type": "task_result", "ok": True, "task": res, "summary": summary})
+
+        elif msg_type == "task_delete":
+            if not self._scheduler:
+                await self._send(ws, {"type": "task_result", "error": "scheduler not running"})
+                return
+            ok, err = self._scheduler.task_delete(msg.get("name", "").strip())
+            if not ok:
+                await self._send(ws, {"type": "task_result", "error": err})
+                return
+            summary = await self._scheduler.apply_tasks(self._scheduler._load_tasks())
+            await self._send(ws, {"type": "task_result", "ok": True, "summary": summary})
+
+        elif msg_type == "task_template_list":
+            if not self._scheduler:
+                await self._send(ws, {"type": "templates_list", "templates": []})
+                return
+            await self._send(ws, {
+                "type": "templates_list",
+                "templates": self._scheduler.list_templates(),
+            })
+
+        elif msg_type == "task_template_create":
+            if not self._scheduler:
+                await self._send(ws, {"type": "template_result", "error": "scheduler not running"})
+                return
+            ok, err = self._scheduler.template_create(
+                msg.get("name", "").strip(), msg.get("prompt", "")
+            )
+            await self._send(ws, {"type": "template_result", "ok": ok, **({"error": err} if not ok else {})})
+
+        elif msg_type == "task_template_update":
+            if not self._scheduler:
+                await self._send(ws, {"type": "template_result", "error": "scheduler not running"})
+                return
+            ok, err = self._scheduler.template_update(
+                msg.get("name", "").strip(), msg.get("prompt", "")
+            )
+            await self._send(ws, {"type": "template_result", "ok": ok, **({"error": err} if not ok else {})})
+
+        elif msg_type == "task_template_delete":
+            if not self._scheduler:
+                await self._send(ws, {"type": "template_result", "error": "scheduler not running"})
+                return
+            ok, err = self._scheduler.template_delete(msg.get("name", "").strip())
+            await self._send(ws, {"type": "template_result", "ok": ok, **({"error": err} if not ok else {})})
+
         elif msg_type == "compact":
             cwd = msg.get("cwd", "")
             session_id = msg.get("session_id", "")
diff --git a/emrg/server/scheduler.py b/emrg/server/scheduler.py
index 40870af..a7fe98c 100644
--- a/emrg/server/scheduler.py
+++ b/emrg/server/scheduler.py
@@ -17,6 +17,7 @@
 import json
 import logging
 import os
+import re
 import subprocess
 import time
 from datetime import datetime
@@ -53,6 +54,57 @@
     "promote": "promote_prompt.md",
 }
 
+# ── Task CRUD constants (rant 2026-08-12T18:23:15 P2) ─────────────
+TASK_NAME_RE = re.compile(r"^[a-z0-9][a-z0-9-]*$")
+TASK_NAME_MAX = 32
+MIN_INTERVAL = 60
+DEFAULT_INTERVAL = 1800
+
+
+def _task_templates_dir() -> Path:
+    """Directory holding user-defined task type templates."""
+    return config_dir() / "task-templates"
+
+
+def _custom_templates() -> list[str]:
+    """Names of user-defined task types (sorted, *.md basenames)."""
+    d = _task_templates_dir()
+    if not d.is_dir():
+        return []
+    try:
+        return sorted(p.stem for p in d.glob("*.md"))
+    except OSError:
+        return []
+
+
+def _read_custom_template(name: str) -> str | None:
+    """Read a user template's prompt text; None if missing."""
+    p = _task_templates_dir() / f"{name}.md"
+    try:
+        if p.exists():
+            return p.read_text(encoding="utf-8")
+    except OSError:
+        pass
+    return None
+
+
+def _write_custom_template(name: str, prompt: str) -> None:
+    """Atomically write a user template."""
+    d = _task_templates_dir()
+    d.mkdir(parents=True, exist_ok=True)
+    p = d / f"{name}.md"
+    tmp = p.with_suffix(".md.tmp")
+    tmp.write_text(prompt, encoding="utf-8")
+    tmp.replace(p)
+
+
+def _delete_custom_template(name: str) -> None:
+    p = _task_templates_dir() / f"{name}.md"
+    try:
+        p.unlink()
+    except OSError:
+        pass
+
 
 def _resolve_task_template(task_type: str) -> Path:
     """Resolve the prompt template for a task type.
@@ -72,6 +124,22 @@ def _resolve_task_template(task_type: str) -> Path:
     return Path(__file__).parent / "evolution_prompt.md"
 
 
+def _task_cfg_signature(cfg: dict) -> tuple:
+    """Stable signature of a task cfg for hot-reload diffing.
+
+    Any change to type / config / interval / enabled marks the task
+    as needing a handler restart.
+    """
+    conf = cfg.get("config") if isinstance(cfg.get("config"), dict) else {}
+    return (
+        cfg.get("name"),
+        cfg.get("type"),
+        json.dumps(conf, sort_keys=True),
+        cfg.get("interval", DEFAULT_INTERVAL),
+        bool(cfg.get("enabled", True)),
+    )
+
+
 def _resolve_project_path(name: str) -> str | None:
     """Resolve a project name to its path from projects.yml."""
     projects_file = config_dir() / "projects.yml"
@@ -1027,6 +1095,37 @@ def __init__(self, identity: InstanceIdentity) -> None:
         self._tasks_file = config_dir() / "tasks.yml"
         self._handlers: list[TaskHandler] = []
         self._coros: list[asyncio.Task] = []
+        # cfg (from tasks.yml) used to start each handler — for hot-reload diffing.
+        self._handler_cfgs: dict[str, dict] = {}
+
+    def _start_handler_for(self, cfg: dict) -> TaskHandler:
+        """Create + start a handler for a task cfg; returns the handler."""
+        template_path = _resolve_task_template(cfg["type"])
+        handler = TaskHandler(
+            name=cfg["name"],
+            config=cfg.get("config", {}),
+            interval=cfg.get("interval", DEFAULT_INTERVAL),
+            identity=self.identity,
+            template_path=template_path,
+        )
+        self._handlers.append(handler)
+        self._handler_cfgs[handler.name] = cfg
+        self._coros.append(asyncio.create_task(handler.run()))
+        return handler
+
+    def _stop_handler(self, handler: TaskHandler) -> None:
+        """Stop a handler and cancel its coroutine (hot-reload removal/restart)."""
+        handler.stop()
+        try:
+            idx = self._handlers.index(handler)
+        except ValueError:
+            idx = -1
+        if idx >= 0:
+            coro = self._coros[idx]
+            coro.cancel()
+            del self._coros[idx]
+            del self._handlers[idx]
+        self._handler_cfgs.pop(handler.name, None)
 
     def load_and_start(self) -> list[asyncio.Task]:
         """Load tasks.yml, start all enabled tasks, return coroutine list."""
@@ -1045,27 +1144,17 @@ def load_and_start(self) -> list[asyncio.Task]:
         for cfg in tasks_config:
             if not cfg.get("enabled", True):
                 continue
-            handler_cls = self.HANDLERS.get(cfg["type"])
+            handler_cls = self.HANDLERS.get(cfg["type"], TaskHandler)
             if handler_cls is None:
                 logger.warning(
                     "TaskScheduler: unknown type %r for task %r",
                     cfg["type"], cfg["name"],
                 )
                 continue
-            template_path = _resolve_task_template(cfg["type"])
-            handler = handler_cls(
-                name=cfg["name"],
-                config=cfg.get("config", {}),
-                interval=cfg.get("interval", 1800),
-                identity=self.identity,
-                template_path=template_path,
-            )
-            self._handlers.append(handler)
-            coro = asyncio.create_task(handler.run())
-            self._coros.append(coro)
+            self._start_handler_for(cfg)
             logger.info(
                 "TaskScheduler: started %s[%s] every %ds",
-                cfg["type"], cfg["name"], cfg.get("interval", 1800),
+                cfg["type"], cfg["name"], cfg.get("interval", DEFAULT_INTERVAL),
             )
 
         return self._coros
@@ -1238,3 +1327,195 @@ def create_task(self, name: str, task_type: str, config: dict, interval: int) ->
         })
         self._save_tasks(tasks)
         logger.info("TaskScheduler: created task %s", name)
+
+    # ── Task CRUD + hot reload (rant 2026-08-12T18:23:15 P2) ────────
+
+    @staticmethod
+    def _validate_task_fields(
+        name: str, task_type: str, project: str, interval: int,
+    ) -> str | None:
+        """Return an error string, or None when the fields are valid."""
+        if not TASK_NAME_RE.match(name) or len(name) > TASK_NAME_MAX:
+            return f"invalid task name {name!r} (^[a-z0-9][a-z0-9-]*$, <= {TASK_NAME_MAX} chars)"
+        if task_type not in TASK_TEMPLATES and task_type not in _custom_templates():
+            return f"unknown task type {task_type!r} (not builtin, no custom template)"
+        if not project or _resolve_project_path(project) is None:
+            return f"project {project!r} is not registered in projects.yml"
+        if not isinstance(interval, int) or isinstance(interval, bool) or interval < MIN_INTERVAL:
+            return f"interval must be an integer >= {MIN_INTERVAL}"
+        return None
+
+    def task_create(
+        self, name: str, task_type: str, project: str,
+        interval: int | None = None, enabled: bool = True,
+        repo: str | None = None, description: str | None = None,
+    ) -> tuple[bool, str | dict]:
+        """Create a task. Returns (ok, error) or (ok, task-dict)."""
+        interval = DEFAULT_INTERVAL if interval is None else interval
+        err = self._validate_task_fields(name, task_type, project, interval)
+        if err:
+            return False, err
+        tasks = self._load_tasks()
+        if any(t.get("name") == name for t in tasks):
+            return False, f"task {name!r} already exists"
+        cfg: dict = {"project": project}
+        if repo:
+            cfg["repo"] = repo
+        task: dict = {
+            "name": name,
+            "type": task_type,
+            "config": cfg,
+            "interval": interval,
+            "enabled": bool(enabled),
+            "last_run": None,
+        }
+        if description:
+            task["description"] = description
+        tasks.append(task)
+        self._save_tasks(tasks)
+        logger.info("TaskScheduler: task %s created (type=%s)", name, task_type)
+        return True, task
+
+    def task_update(self, name: str, **fields) -> tuple[bool, str | dict]:
+        """Update a task's fields. Returns (ok, error) or (ok, task-dict)."""
+        tasks = self._load_tasks()
+        task = next((t for t in tasks if t.get("name") == name), None)
+        if task is None:
+            return False, f"task {name!r} not found"
+        new_type = fields.get("type", task.get("type", "evolution"))
+        new_project = fields.get("project")
+        if new_project is None:
+            cfg = task.get("config") if isinstance(task.get("config"), dict) else {}
+            new_project = cfg.get("project", "")
+        new_interval = fields.get("interval", task.get("interval", DEFAULT_INTERVAL))
+        err = self._validate_task_fields(name, new_type, new_project, new_interval)
+        if err:
+            return False, err
+        if "type" in fields:
+            task["type"] = fields["type"]
+        if "project" in fields or "repo" in fields:
+            cfg = task.get("config") if isinstance(task.get("config"), dict) else {}
+            if "project" in fields:
+                cfg["project"] = fields["project"]
+            if "repo" in fields:
+                if fields["repo"]:
+                    cfg["repo"] = fields["repo"]
+                else:
+                    cfg.pop("repo", None)
+            task["config"] = cfg
+        if "interval" in fields:
+            task["interval"] = fields["interval"]
+        if "enabled" in fields:
+            task["enabled"] = bool(fields["enabled"])
+        if "description" in fields:
+            task["description"] = fields["description"]
+        self._save_tasks(tasks)
+        logger.info("TaskScheduler: task %s updated", name)
+        return True, task
+
+    def task_delete(self, name: str) -> tuple[bool, str]:
+        """Delete a task by name. Returns (ok, error)."""
+        tasks = self._load_tasks()
+        before = len(tasks)
+        tasks = [t for t in tasks if t.get("name") != name]
+        if len(tasks) == before:
+            return False, f"task {name!r} not found"
+        self._save_tasks(tasks)
+        logger.info("TaskScheduler: task %s deleted", name)
+        return True, ""
+
+    async def apply_tasks(self, tasks: list[dict]) -> dict:
+        """Hot-reload tasks from a new config list (rant 2026-08-12T18:23:15 P2).
+
+        Writes tasks.yml atomically, diffs against running handlers, and
+        starts/stops/restarts handlers as needed — no daemon restart.
+        Idempotent; returns {"added": [...], "removed": [...], "updated": [...]}.
+        """
+        self._save_tasks(tasks)
+        enabled = {
+            t.get("name"): t for t in tasks
+            if isinstance(t, dict) and t.get("enabled", True)
+        }
+        current = {h.name: h for h in list(self._handlers)}
+        added: list[str] = []
+        removed: list[str] = []
+        updated: list[str] = []
+        for name, handler in list(current.items()):
+            if name not in enabled:
+                removed.append(name)
+                self._stop_handler(handler)
+        for name, cfg in enabled.items():
+            if name not in current:
+                added.append(name)
+                self._start_handler_for(cfg)
+            else:
+                old = self._handler_cfgs.get(name)
+                if old is not None and _task_cfg_signature(old) != _task_cfg_signature(cfg):
+                    updated.append(name)
+                    self._stop_handler(current[name])
+                    self._start_handler_for(cfg)
+        return {"added": added, "removed": removed, "updated": updated}
+
+    def list_templates(self) -> list[dict]:
+        """List all task types: builtin (read-only) + custom (with prompt preview)."""
+        result: list[dict] = []
+        for name in sorted(TASK_TEMPLATES):
+            result.append({
+                "name": name,
+                "builtin": True,
+                "template": TASK_TEMPLATES[name],
+            })
+        for name in _custom_templates():
+            result.append({
+                "name": name,
+                "builtin": False,
+                "template": f"{name}.md",
+                "prompt": _read_custom_template(name) or "",
+            })
+        return result
+
+    def template_create(self, name: str, prompt: str) -> tuple[bool, str]:
+        """Create a custom task type template. Returns (ok, error)."""
+        if not TASK_NAME_RE.match(name) or len(name) > TASK_NAME_MAX:
+            return False, f"invalid template name {name!r} (^[a-z0-9][a-z0-9-]*$, <= {TASK_NAME_MAX} chars)"
+        if name in TASK_TEMPLATES:
+            return False, f"builtin task type {name!r} is read-only"
+        if not prompt or not prompt.strip():
+            return False, "template prompt must not be empty"
+        if _read_custom_template(name) is not None:
+            return False, f"template {name!r} already exists"
+        _write_custom_template(name, prompt)
+        logger.info("TaskScheduler: custom template %s created", name)
+        return True, ""
+
+    def template_update(self, name: str, prompt: str) -> tuple[bool, str]:
+        """Update a custom task type template. Returns (ok, error)."""
+        if name in TASK_TEMPLATES:
+            return False, f"builtin task type {name!r} is read-only"
+        if _read_custom_template(name) is None:
+            return False, f"template {name!r} not found"
+        if not prompt or not prompt.strip():
+            return False, "template prompt must not be empty"
+        _write_custom_template(name, prompt)
+        logger.info("TaskScheduler: custom template %s updated", name)
+        return True, ""
+
+    def template_delete(self, name: str) -> tuple[bool, str]:
+        """Delete a custom task type template. Returns (ok, error).
+
+        Refuses when tasks reference the type (host decision, rant 18:23:15).
+        """
+        if name in TASK_TEMPLATES:
+            return False, f"builtin task type {name!r} is read-only"
+        if _read_custom_template(name) is None:
+            return False, f"template {name!r} not found"
+        tasks = self._load_tasks()
+        refs = [t.get("name") for t in tasks if t.get("type") == name]
+        if refs:
+            return False, (
+                f"cannot delete type {name!r}: {len(refs)} task(s) use it "
+                f"({', '.join(str(r) for r in refs[:5])})"
+            )
+        _delete_custom_template(name)
+        logger.info("TaskScheduler: custom template %s deleted", name)
+        return True, ""
diff --git a/tests/test_scheduler.py b/tests/test_scheduler.py
index 9a15cfc..977856c 100644
--- a/tests/test_scheduler.py
+++ b/tests/test_scheduler.py
@@ -320,7 +320,13 @@ async def _run():
 
 
 def test_load_and_start_unknown_type(tmp_path):
-    """Tasks with unknown handler type are skipped; self-heal still adds emrg-task."""
+    """Custom/unknown types start as TaskHandler with the fallback template (P2).
+
+    rant 2026-08-12T18:23:15: types not in TASK_TEMPLATES are treated as
+    user-defined task types — resolved via ~/.emrg/task-templates/<type>.md,
+    falling back to evolution_prompt.md with a warning. Self-heal still adds
+    emrg-task.
+    """
     tasks_yml = tmp_path / "tasks.yml"
     tasks_yml.write_text(yaml.safe_dump([
         {"name": "bad", "type": "nonexistent_handler", "config": {}, "enabled": True},
@@ -339,8 +345,9 @@ async def _run():
     finally:
         mod.config_dir = orig_config
 
-    assert len(coros) == 1  # the self-healed emrg-task
-    assert sched._handlers[0].name == "emrg-task"
+    by_name = {h.name: h for h in sched._handlers}
+    assert set(by_name) == {"bad", "emrg-task"}  # custom type + self-healed
+    assert by_name["bad"]._template_path.name == "evolution_prompt.md"  # fallback
     sched.stop_all()
     for c in coros:
         c.cancel()
@@ -1472,3 +1479,238 @@ def test_remote_advanced_false_without_git_repo(tmp_path):
         assert handler._remote_advanced() is False
     finally:
         mod.subprocess.run = orig_run
+
+
+# ── Task CRUD + hot reload + templates (rant 2026-08-12T18:23:15 P2) ──
+
+
+def _p2_env(tmp_path):
+    """Point config_dir at tmp_path with a registered project."""
+    from emrg.server import scheduler as mod
+    projects_yml = tmp_path / "projects.yml"
+    projects_yml.write_text(yaml.safe_dump([
+        {"name": "emrg", "path": str(tmp_path / "emrg")},
+        {"name": "mem", "path": str(tmp_path / "mem")},
+    ]))
+    orig = mod.config_dir
+    mod.config_dir = lambda: tmp_path
+    return mod, orig
+
+
+def test_task_create_validation(tmp_path):
+    """Invalid name / unknown type / unregistered project / interval<60 rejected."""
+    from emrg.server import scheduler as mod
+    mod, orig = _p2_env(tmp_path)
+    try:
+        sched = TaskScheduler(InstanceIdentity())
+        sched._tasks_file = tmp_path / "tasks.yml"
+        ok, err = sched.task_create("Bad Name", "evolution", "emrg", 60)
+        assert not ok and "invalid task name" in err
+        ok, err = sched.task_create("good", "no-such-type", "emrg", 60)
+        assert not ok and "unknown task type" in err
+        ok, err = sched.task_create("good", "evolution", "not-registered", 60)
+        assert not ok and "not registered" in err
+        ok, err = sched.task_create("good", "evolution", "emrg", 30)
+        assert not ok and ">= 60" in err
+        ok, err = sched.task_create("good", "evolution", "emrg", "abc")
+        assert not ok and ">= 60" in err
+    finally:
+        mod.config_dir = orig
+
+
+def test_task_create_and_duplicate(tmp_path):
+    """Valid task create persists to tasks.yml; duplicate rejected."""
+    from emrg.server import scheduler as mod
+    mod, orig = _p2_env(tmp_path)
+    try:
+        sched = TaskScheduler(InstanceIdentity())
+        tasks_file = tmp_path / "tasks.yml"
+        sched._tasks_file = tasks_file
+        ok, task = sched.task_create("daily", "evolution", "mem", 300, repo="acme/x")
+        assert ok and task["name"] == "daily" and task["interval"] == 300
+        assert task["config"] == {"project": "mem", "repo": "acme/x"}
+        saved = yaml.safe_load(tasks_file.read_text(encoding="utf-8"))
+        assert any(t["name"] == "daily" for t in saved)
+        ok, err = sched.task_create("daily", "evolution", "emrg", 60)
+        assert not ok and "already exists" in err
+    finally:
+        mod.config_dir = orig
+
+
+def test_task_update_and_delete(tmp_path):
+    """Update changes fields; delete removes the entry; not-found errors."""
+    from emrg.server import scheduler as mod
+    mod, orig = _p2_env(tmp_path)
+    try:
+        sched = TaskScheduler(InstanceIdentity())
+        sched._tasks_file = tmp_path / "tasks.yml"
+        sched.task_create("daily", "evolution", "mem", 300)
+        ok, task = sched.task_update("daily", interval=600, enabled=False, repo="acme/y")
+        assert ok and task["interval"] == 600 and task["enabled"] is False
+        assert task["config"]["repo"] == "acme/y"
+        ok, err = sched.task_update("nope", interval=60)
+        assert not ok and "not found" in err
+        ok, err = sched.task_delete("daily")
+        assert ok and err == ""
+        ok, err = sched.task_delete("daily")
+        assert not ok and "not found" in err
+    finally:
+        mod.config_dir = orig
+
+
+def test_apply_tasks_hot_reload(tmp_path):
+    """apply_tasks diffs handlers: add / remove / restart on change (no daemon restart)."""
+    from emrg.server import scheduler as mod
+    mod, orig = _p2_env(tmp_path)
+    try:
+        sched = TaskScheduler(InstanceIdentity())
+        sched._tasks_file = tmp_path / "tasks.yml"
+        sched._save_tasks([
+            {"name": "a", "type": "evolution", "config": {"project": "emrg"}, "interval": 300, "enabled": True},
+            {"name": "b", "type": "evolution", "config": {"project": "mem"}, "interval": 300, "enabled": True},
+        ])
+
+        async def _load_and_diff(new_tasks):
+            sched.load_and_start()
+            assert {h.name for h in sched._handlers} == {"a", "b"}
+            summary = await sched.apply_tasks(new_tasks)
+            return summary
+
+        summary = asyncio.run(_load_and_diff([
+            {"name": "a", "type": "evolution", "config": {"project": "emrg"}, "interval": 300, "enabled": True},
+            {"name": "c", "type": "evolution", "config": {"project": "mem"}, "interval": 900, "enabled": True},
+        ]))
+        assert summary["removed"] == ["b"]
+        assert summary["added"] == ["c"]
+        assert summary["updated"] == []
+        names = {h.name for h in sched._handlers}
+        assert names == {"a", "c"}
+        c = next(h for h in sched._handlers if h.name == "c")
+        assert c.interval == 900
+        sched.stop_all()
+    finally:
+        mod.config_dir = orig
+
+
+def test_apply_tasks_update_restart(tmp_path):
+    """Changing a task's interval restarts (stops + starts) its handler."""
+    from emrg.server import scheduler as mod
+    mod, orig = _p2_env(tmp_path)
+    try:
+        sched = TaskScheduler(InstanceIdentity())
+        sched._tasks_file = tmp_path / "tasks.yml"
+
+        async def _run():
+            sched._save_tasks([
+                {"name": "a", "type": "evolution", "config": {"project": "emrg"}, "interval": 300, "enabled": True},
+            ])
+            sched.load_and_start()
+            h_old = sched._handlers[0]
+            summary = await sched.apply_tasks([
+                {"name": "a", "type": "evolution", "config": {"project": "emrg"}, "interval": 600, "enabled": True},
+            ])
+            return h_old, summary
+
+        h_old, summary = asyncio.run(_run())
+        assert summary["updated"] == ["a"]
+        assert summary["added"] == [] and summary["removed"] == []
+        assert len(sched._handlers) == 1
+        assert sched._handlers[0] is not h_old  # restarted
+        assert sched._handlers[0].interval == 600
+        sched.stop_all()
+    finally:
+        mod.config_dir = orig
+
+
+def test_apply_tasks_idempotent(tmp_path):
+    """Applying the same tasks is a no-op (no add/remove/update)."""
+    from emrg.server import scheduler as mod
+    mod, orig = _p2_env(tmp_path)
+    try:
+        sched = TaskScheduler(InstanceIdentity())
+        sched._tasks_file = tmp_path / "tasks.yml"
+        tasks = [
+            {"name": "a", "type": "evolution", "config": {"project": "emrg"}, "interval": 300, "enabled": True},
+        ]
+
+        async def _run():
+            sched._save_tasks(tasks)
+            sched.load_and_start()
+            return await sched.apply_tasks(tasks)
+
+        summary = asyncio.run(_run())
+        assert summary == {"added": [], "removed": [], "updated": []}
+        assert len(sched._handlers) == 1
+        sched.stop_all()
+    finally:
+        mod.config_dir = orig
+
+
+def test_template_crud_and_guards(tmp_path):
+    """Custom templates: create/list/update/delete; builtin read-only; delete-refused guard."""
+    from emrg.server import scheduler as mod
+    mod, orig = _p2_env(tmp_path)
+    try:
+        sched = TaskScheduler(InstanceIdentity())
+        sched._tasks_file = tmp_path / "tasks.yml"
+        # builtin read-only
+        ok, err = sched.template_create("evolution", "x")
+        assert not ok and "read-only" in err
+        ok, err = sched.template_update("evolution", "x")
+        assert not ok and "read-only" in err
+        ok, err = sched.template_delete("evolution")
+        assert not ok and "read-only" in err
+        # create
+        ok, err = sched.template_create("report", "# Report {{ instance_id }}")
+        assert ok and err == ""
+        ok, err = sched.template_create("report", "dup")
+        assert not ok and "already exists" in err
+        ok, err = sched.template_create("Bad Name", "x")
+        assert not ok and "invalid template name" in err
+        ok, err = sched.template_create("empty", "   ")
+        assert not ok and "must not be empty" in err
+        # list
+        templates = {t["name"]: t for t in sched.list_templates()}
+        assert templates["evolution"]["builtin"] is True
+        assert templates["report"]["builtin"] is False
+        assert "instance_id" in templates["report"]["prompt"]
+        # update
+        ok, err = sched.template_update("report", "# New")
+        assert ok
+        assert mod._read_custom_template("report") == "# New"
+        ok, err = sched.template_update("missing", "x")
+        assert not ok and "not found" in err
+        # delete referenced → refused (host decision)
+        sched.task_create("uses-report", "report", "mem", 300)
+        ok, err = sched.template_delete("report")
+        assert not ok and "1 task(s) use it" in err
+        # delete after removing reference → ok
+        sched.task_delete("uses-report")
+        ok, err = sched.template_delete("report")
+        assert ok and err == ""
+        assert mod._read_custom_template("report") is None
+    finally:
+        mod.config_dir = orig
+
+
+def test_task_create_custom_type(tmp_path):
+    """A custom template type can be used to create a runnable task."""
+    from emrg.server import scheduler as mod
+    mod, orig = _p2_env(tmp_path)
+    try:
+        sched = TaskScheduler(InstanceIdentity())
+        sched._tasks_file = tmp_path / "tasks.yml"
+        sched.template_create("report", "# Report {{ instance_id }}")
+
+        async def _run():
+            ok, task = sched.task_create("daily-report", "report", "mem", 300)
+            assert ok and task["type"] == "report"
+            sched._save_tasks([task])
+            sched.load_and_start()
+            h = next(h for h in sched._handlers if h.name == "daily-report")
+            assert h._template_path == tmp_path / "task-templates" / "report.md"
+            sched.stop_all()
+
+        asyncio.run(_run())
+    finally:
+        mod.config_dir = orig