diff --git a/.agents/skills/lib.sh b/.agents/skills/lib.sh
index 0959389..298cf92 100644
--- a/.agents/skills/lib.sh
+++ b/.agents/skills/lib.sh
@@ -1873,7 +1873,7 @@ _delegate_script() {
 # delegate_submit_job <prompt> <agent> <agent_session>
 #
 # Register a job in the multi-agent-mux-delegate-job registry. Prints the new JID on stdout.
-delegate_submit_job() {
+_delegate_submit_job_legacy() {
   local prompt="$1" agent="$2" session="$3"
   local py_bin registry_py
   py_bin="$(_delegate_py_bin)"
@@ -1894,7 +1894,7 @@ delegate_submit_job() {
 # inline .venv-walk + publish_event.py blocks that were duplicated across
 # create/delete/resume (review item 7). Non-fatal by contract: an empty job id,
 # a missing script, or a broker failure never aborts the caller.
-delegate_publish_event() {
+_delegate_publish_event_legacy() {
   local job_id="$1" event="$2" detail="${3:-}"
   [ -n "$job_id" ] || return 0
   local py_bin pub
@@ -1909,7 +1909,7 @@ delegate_publish_event() {
 # The watchdog re-spawns the subscriber every 2 minutes (or whatever hard
 # limit we set) and exits automatically when the JOB reaches terminal state.
 # Returns the watchdog PID via stdout.
-start_watchdog() {
+_start_watchdog_legacy() {
   local job_id="$1"
   local workdir="${2:-$PWD}"
   local monitor_script="$workdir/.agents/skills/multi-agent-mux-monitor/scripts/reconcile.sh"
@@ -1937,6 +1937,59 @@ start_watchdog() {
   echo "$pid"
 }
 
+# ---------------------------------------------------------------------------
+# Python-backed delegation helpers (issue #5 phase 1).
+#
+# The bodies above are kept verbatim as _*_legacy and are still reachable with
+# MAM_DELEGATION_PY=0, which is the instant rollback switch: no redeploy, no
+# revert. They are removed in phase 4, once the python path has soaked.
+#
+# PYTHONPATH is passed explicitly (same pattern as env_python) so `-m` resolves
+# lib_py from the tree this lib.sh belongs to -- that is what makes a frozen
+# loop run frozen delegation code.
+# ---------------------------------------------------------------------------
+
+# delegate_submit_job <prompt> <agent> <agent_session>
+# Register a job. Prints the new JID on stdout and nothing else.
+delegate_submit_job() {
+  if [ "${MAM_DELEGATION_PY:-1}" = "0" ]; then _delegate_submit_job_legacy "$@"; return $?; fi
+  local prompt="$1" agent="$2" session="$3"
+  local registry_py
+  registry_py="$(_delegate_script registry.py)"
+  if [ -z "$registry_py" ] || [ ! -f "$registry_py" ]; then
+    echo "ERROR: multi-agent-mux-delegate-job registry.py not found under .agents/skills/" >&2
+    return 1
+  fi
+  PYTHONPATH="$SKILL_DIR:${PYTHONPATH:-}" "$(_delegate_py_bin)" -m lib_py.delegation.cli \
+    submit --prompt "$prompt" --agent "$agent" --agent-session "$session"
+}
+
+# delegate_publish_event <job_id> <event> [detail]
+# Non-fatal by contract: an empty job id, a missing script tree, or a broker
+# failure must never abort the caller.
+delegate_publish_event() {
+  local job_id="$1" event="$2" detail="${3:-}"
+  [ -n "$job_id" ] || return 0
+  if [ "${MAM_DELEGATION_PY:-1}" = "0" ]; then _delegate_publish_event_legacy "$@"; return $?; fi
+  local pub
+  pub="$(_delegate_script publish_event.py)"
+  [ -n "$pub" ] && [ -f "$pub" ] || return 0
+  PYTHONPATH="$SKILL_DIR:${PYTHONPATH:-}" "$(_delegate_py_bin)" -m lib_py.delegation.cli \
+    publish --job "$job_id" --event "$event" --detail "$detail" || true
+}
+
+# ensure_subscriber_daemon <job_id> [workdir]
+# Start the monitor subscriber for this workspace unless one is already running.
+# Prints the PID(s) on stdout, exactly as start_watchdog always did.
+ensure_subscriber_daemon() {
+  if [ "${MAM_DELEGATION_PY:-1}" = "0" ]; then _start_watchdog_legacy "$@"; return $?; fi
+  PYTHONPATH="$SKILL_DIR:${PYTHONPATH:-}" "$(_delegate_py_bin)" -m lib_py.delegation.cli \
+    ensure-daemon --job "$1" --workdir "${2:-$PWD}"
+}
+
+# start_watchdog — retained name; create_session.sh and operator docs use it.
+start_watchdog() { ensure_subscriber_daemon "$@"; }
+
 # wait_for_tui_ready <session_name> <agent>
 # Waits up to 30 seconds for the agent's TUI to render its welcome screen.
 # Returns 0 on readiness, 1 on timeout, 2 on unobservable/headless.
diff --git a/.agents/skills/lib_py/delegation/__init__.py b/.agents/skills/lib_py/delegation/__init__.py
index f5a488d..383da88 100644
--- a/.agents/skills/lib_py/delegation/__init__.py
+++ b/.agents/skills/lib_py/delegation/__init__.py
@@ -16,3 +16,7 @@ Package rules, enforced by tests:
     sibling import would silently mix trees.
 """
 
+from lib_py.delegation.daemon import ensure_subscriber_daemon  # noqa: E402,F401
+from lib_py.delegation.service import DelegationService        # noqa: E402,F401
+
+__all__ = ["DelegationService", "ensure_subscriber_daemon"]
diff --git a/.agents/skills/lib_py/delegation/__main__.py b/.agents/skills/lib_py/delegation/__main__.py
new file mode 100644
index 0000000..25a8b45
--- /dev/null
+++ b/.agents/skills/lib_py/delegation/__main__.py
@@ -0,0 +1,7 @@
+"""Allow `python -m lib_py.delegation` as a synonym for the cli router."""
+import sys
+
+from lib_py.delegation.cli import main
+
+if __name__ == "__main__":
+    sys.exit(main())
diff --git a/.agents/skills/lib_py/delegation/cli.py b/.agents/skills/lib_py/delegation/cli.py
new file mode 100644
index 0000000..06af9c3
--- /dev/null
+++ b/.agents/skills/lib_py/delegation/cli.py
@@ -0,0 +1,90 @@
+"""`python -m lib_py.delegation.cli` — the router the shell helpers call.
+
+Deliberately thin: every subcommand delegates to the existing `main()` of the
+module that already owned that behaviour, so exit codes, stdout shape and side
+effects cannot drift between the shell path and the script path. The one piece
+of real logic here is `ensure-daemon`, which reproduces the shell helper's
+stderr wording and exit status for a missing monitor script.
+"""
+from __future__ import annotations
+
+import argparse
+import logging
+import sys
+from typing import List, Optional
+
+from lib_py.delegation import daemon, mqtt_common, publish_event, registry
+
+logger = logging.getLogger("delegate_job.cli")
+
+
+def _build_parser() -> argparse.ArgumentParser:
+    parser = argparse.ArgumentParser(
+        prog="lib_py.delegation.cli",
+        description="delegation helpers invoked by lib.sh",
+    )
+    sub = parser.add_subparsers(dest="command", required=True)
+
+    p_submit = sub.add_parser("submit", help="register a job; prints the job id")
+    p_submit.add_argument("--prompt", required=True)
+    p_submit.add_argument("--agent", default="claude-code")
+    p_submit.add_argument("--agent-session", default="herdr:claude")
+    p_submit.add_argument("--registry-dir", default=None)
+
+    p_pub = sub.add_parser("publish", help="publish a job lifecycle event")
+    p_pub.add_argument("--job", required=True)
+    p_pub.add_argument("--event", default="progress", choices=publish_event.VALID_EVENTS)
+    p_pub.add_argument("--detail", default="")
+    p_pub.add_argument("--data", default=None)
+    p_pub.add_argument("--retained", action="store_true")
+    p_pub.add_argument("--registry-dir", default=None)
+    p_pub.add_argument("--attempts", type=int, default=3)
+    p_pub.add_argument("-v", "--verbose", action="store_true")
+
+    p_daemon = sub.add_parser("ensure-daemon", help="start the subscriber daemon if absent")
+    p_daemon.add_argument("--job", required=True)
+    p_daemon.add_argument("--workdir", default=None)
+
+    return parser
+
+
+def main(argv: Optional[List[str]] = None) -> int:
+    mqtt_common.ensure_env_loaded()
+    args = _build_parser().parse_args(argv)
+
+    if args.command == "submit":
+        forward = ["register", "--prompt", args.prompt,
+                   "--agent", args.agent, "--agent-session", args.agent_session]
+        if args.registry_dir:
+            forward = ["--registry-dir", args.registry_dir] + forward
+        return registry.main(forward)
+
+    if args.command == "publish":
+        forward = ["--job", args.job, "--event", args.event,
+                   "--detail", args.detail, "--attempts", str(args.attempts)]
+        if args.data is not None:
+            forward += ["--data", args.data]
+        if args.retained:
+            forward += ["--retained"]
+        if args.registry_dir:
+            forward += ["--registry-dir", args.registry_dir]
+        if args.verbose:
+            forward += ["--verbose"]
+        return publish_event.main(forward)
+
+    if args.command == "ensure-daemon":
+        try:
+            pids = daemon.ensure_subscriber_daemon(args.job, args.workdir)
+        except FileNotFoundError as exc:
+            # Byte-identical to the legacy shell helper's failure path.
+            print("ERROR: monitor script not found: {}".format(exc.args[0]), file=sys.stderr)
+            return 1
+        for pid in pids:
+            print(pid)
+        return 0
+
+    return 1
+
+
+if __name__ == "__main__":
+    sys.exit(main())
diff --git a/.agents/skills/lib_py/delegation/daemon.py b/.agents/skills/lib_py/delegation/daemon.py
new file mode 100644
index 0000000..54b79d8
--- /dev/null
+++ b/.agents/skills/lib_py/delegation/daemon.py
@@ -0,0 +1,86 @@
+"""Subscriber daemon supervision — the python port of lib.sh's `start_watchdog`.
+
+Behaviour is byte-for-byte compatible with the shell original (issue #5 phase 1,
+D5): same monitor path, same stderr wording and exit status when it is missing,
+same `pgrep -f` pre-check pattern, same log file, same "print the PID" contract.
+Callers capture stdout with $( ), so nothing else may be printed there.
+"""
+from __future__ import annotations
+
+import logging
+import os
+import signal
+import subprocess
+from typing import List, Optional
+
+logger = logging.getLogger("delegate_job.daemon")
+
+MONITOR_RELPATH = os.path.join(
+    ".agents", "skills", "multi-agent-mux-monitor", "scripts", "reconcile.sh"
+)
+LOG_RELPATH = os.path.join(".mam", "multi-agent-mux-monitor.log")
+
+
+def monitor_script_path(workdir: str) -> str:
+    return os.path.join(workdir, MONITOR_RELPATH)
+
+
+def find_running(monitor_script: str) -> List[str]:
+    """PIDs of subscribers already running for this monitor script.
+
+    Uses the exact `pgrep -f` pattern the shell helper used, so a daemon started
+    by either implementation is found by the other. Multiple matches are returned
+    in full: the legacy helper echoed every line and callers only ever echo it
+    back, so trimming here would be a silent behaviour change.
+    """
+    try:
+        res = subprocess.run(
+            ["pgrep", "-f", "bash {} --subscribe".format(monitor_script)],
+            capture_output=True, text=True,
+        )
+    except OSError as exc:  # pragma: no cover - pgrep missing
+        logger.warning("pgrep unavailable: %s", exc)
+        return []
+    return [line for line in res.stdout.split() if line.strip()]
+
+
+def _ignore_sighup():  # pragma: no cover - runs in the forked child
+    """Match `nohup`: SIG_IGN survives execve, so the daemon outlives its parent."""
+    signal.signal(signal.SIGHUP, signal.SIG_IGN)
+
+
+def spawn(monitor_script: str, workdir: str) -> int:
+    log_file = os.path.join(workdir, LOG_RELPATH)
+    os.makedirs(os.path.dirname(log_file), exist_ok=True)
+    log_handle = open(log_file, "ab")
+    try:
+        proc = subprocess.Popen(
+            ["bash", monitor_script, "--subscribe", "--idle-timeout", "0"],
+            cwd=workdir,
+            stdin=subprocess.DEVNULL,
+            stdout=log_handle,
+            stderr=subprocess.STDOUT,
+            start_new_session=True,
+            preexec_fn=_ignore_sighup,
+        )
+    finally:
+        log_handle.close()
+    return proc.pid
+
+
+def ensure_subscriber_daemon(job_id: str, workdir: Optional[str] = None) -> List[str]:
+    """Return the PIDs of the running subscriber, starting one if needed.
+
+    `job_id` is accepted for signature compatibility and is deliberately unused:
+    the daemon is per-workspace (log paths anchor on `workdir`), exactly as in
+    the shell original. Raises FileNotFoundError when the monitor is absent so
+    the CLI can reproduce the legacy stderr line and exit 1.
+    """
+    workdir = workdir or os.getcwd()
+    monitor_script = monitor_script_path(workdir)
+    if not os.path.isfile(monitor_script):
+        raise FileNotFoundError(monitor_script)
+    running = find_running(monitor_script)
+    if running:
+        return running
+    return [str(spawn(monitor_script, workdir))]
diff --git a/.agents/skills/lib_py/delegation/service.py b/.agents/skills/lib_py/delegation/service.py
new file mode 100644
index 0000000..449a480
--- /dev/null
+++ b/.agents/skills/lib_py/delegation/service.py
@@ -0,0 +1,67 @@
+"""DelegationService — one façade over registering, publishing and supervising.
+
+Phase 1 keeps every behaviour in the existing entry points; this only groups
+them so a long-running caller (the phase 2a engine) has a single object to hold,
+optionally with a persistent MQTT connection. CLI and shell callers always use
+`persistent=False`, because a process that exits after one publish gains nothing
+from connection reuse.
+"""
+from __future__ import annotations
+
+import logging
+from typing import Any, Dict, List, Optional
+
+from lib_py.delegation import daemon, mqtt_common, publish_event, registry
+
+logger = logging.getLogger("delegate_job.service")
+
+
+class DelegationService(object):
+    def __init__(self, registry_dir=None, persistent=False):
+        self.registry_dir = registry_dir or mqtt_common.DEFAULT_REGISTRY_DIR
+        self.persistent = persistent
+        self._publisher = None
+
+    # -- lifecycle ---------------------------------------------------------
+    def __enter__(self):
+        if self.persistent:
+            from lib_py.delegation.publisher import PersistentPublisher
+            self._publisher = PersistentPublisher()
+            self._publisher.open()
+        return self
+
+    def __exit__(self, exc_type, exc, tb):
+        self.close()
+        return False
+
+    def close(self):
+        if self._publisher is not None:
+            self._publisher.close()
+            self._publisher = None
+
+    # -- operations --------------------------------------------------------
+    def submit_job(self, prompt, agent="claude-code", agent_session="herdr:claude", **kw):
+        return registry.register_job(
+            prompt=prompt, agent=agent, agent_session=agent_session,
+            registry_dir=self.registry_dir, **kw
+        )
+
+    def publish_event(self, job_id, event, detail="", data=None, attempts=3):
+        """Publish one event. Returns the same exit status as the CLI (0 / 2)."""
+        prepared = publish_event.prepare_event(
+            job_id=job_id, event=event, detail=detail,
+            data=data, registry_dir=self.registry_dir,
+        )
+        publisher = self._publisher
+        if publisher is None:
+            from lib_py.delegation.publisher import OneShotPublisher
+            publisher = OneShotPublisher()
+        ok, error = publisher.publish_with_retry(prepared, attempts=attempts)
+        publish_event.record_outcome(prepared, self.registry_dir, ok, error)
+        return 0 if ok else 2
+
+    def ensure_subscriber_daemon(self, job_id, workdir=None):
+        return daemon.ensure_subscriber_daemon(job_id, workdir)
+
+    def load_job(self, job_id) -> Dict[str, Any]:
+        return mqtt_common.load_job(job_id, self.registry_dir)
diff --git a/tests/test_issue5_phase1_delegation.py b/tests/test_issue5_phase1_delegation.py
index 000a75d..0abdf13 100644
--- a/tests/test_issue5_phase1_delegation.py
+++ b/tests/test_issue5_phase1_delegation.py
@@ -607,3 +607,116 @@ def test_k6_subprocess_executes_the_copy_it_was_given(shim_trees):
     assert probe.returncode == 0, probe.stderr
     assert str(shim_trees / "sandbox") in probe.stdout, probe.stdout
 
+
+# ==============================================================================
+# 5. P1-02 — daemon module, cli router, shell-to-python equivalence
+# ==============================================================================
+
+def _delegation_env(mam_sandbox):
+    """Env for running the package out of the sandbox copy."""
+    env = dict(os.environ)
+    env["PYTHONPATH"] = str(_sandbox_skills(mam_sandbox))
+    env["MAM_ENV_FILE"] = str(Path(mam_sandbox) / ".mam.env.absent")
+    return env
+
+
+def test_daemon_missing_monitor_raises_with_the_legacy_path(mam_sandbox):
+    from lib_py.delegation import daemon
+
+    workdir = Path(mam_sandbox) / "no_monitor"
+    workdir.mkdir()
+    with pytest.raises(FileNotFoundError) as exc:
+        daemon.ensure_subscriber_daemon("job1", str(workdir))
+    assert exc.value.args[0] == str(daemon.monitor_script_path(str(workdir)))
+
+
+def test_daemon_spawn_dedup_and_survives_parent_exit(mam_sandbox):
+    """The daemon must outlive the shell that started it and never double-spawn."""
+    from lib_py.delegation import daemon
+
+    monitor = _install_fake_monitor(mam_sandbox)
+    workdir = str(mam_sandbox)
+    spawned = []
+    try:
+        first = daemon.ensure_subscriber_daemon("job1", workdir)
+        spawned += first
+        assert len(first) == 1 and _alive(first[0])
+        assert (Path(workdir) / ".mam" / "multi-agent-mux-monitor.log").exists()
+        second = daemon.ensure_subscriber_daemon("job1", workdir)
+        assert second == first, "a running daemon must be reused, not replaced"
+        # detached into its own session, so a parent's process-group signal misses it
+        pgid = os.getpgid(int(first[0]))
+        assert pgid != os.getpgid(os.getpid())
+        assert daemon.find_running(str(monitor)) == first
+    finally:
+        _kill(spawned)
+
+
+def test_cli_ensure_daemon_reproduces_the_legacy_failure_line(mam_sandbox):
+    workdir = Path(mam_sandbox) / "no_monitor_cli"
+    workdir.mkdir()
+    res = subprocess.run(
+        [sys.executable, "-m", "lib_py.delegation.cli",
+         "ensure-daemon", "--job", "j1", "--workdir", str(workdir)],
+        capture_output=True, text=True, env=_delegation_env(mam_sandbox), timeout=60,
+    )
+    assert res.returncode == 1
+    expected = str(workdir / ".agents" / "skills" / "multi-agent-mux-monitor"
+                   / "scripts" / "reconcile.sh")
+    assert res.stderr.strip() == "ERROR: monitor script not found: {}".format(expected)
+    assert res.stdout == ""
+
+
+def test_cli_submit_prints_only_the_job_id(mam_sandbox):
+    reg_dir = Path(mam_sandbox) / ".mam" / "jobs"
+    reg_dir.mkdir(parents=True, exist_ok=True)
+    res = subprocess.run(
+        [sys.executable, "-m", "lib_py.delegation.cli", "submit",
+         "--prompt", "hello", "--agent", "claude-code",
+         "--agent-session", "herdr:claude", "--registry-dir", str(reg_dir)],
+        capture_output=True, text=True, env=_delegation_env(mam_sandbox), timeout=60,
+    )
+    assert res.returncode == 0, res.stderr
+    job_id = res.stdout.strip()
+    assert re.fullmatch(r"[0-9a-f]+", job_id), repr(res.stdout)
+    assert res.stdout == job_id + "\n", "stdout must carry the id and nothing else"
+    assert (reg_dir / (job_id + ".json")).exists()
+
+
+@pytest.mark.parametrize("entry", ["script", "cli"])
+def test_publish_failure_semantics_match_across_entry_points(mam_sandbox, entry):
+    """G-1/G-2/G-4 semantics must be identical through the shim and through the cli.
+
+    Broker points at a non-routable port, so nothing touches the network."""
+    reg_dir = Path(mam_sandbox) / ".mam" / "jobs"
+    reg_dir.mkdir(parents=True, exist_ok=True)
+    env = _delegation_env(mam_sandbox)
+    job_id = subprocess.run(
+        [sys.executable, "-m", "lib_py.delegation.cli", "submit", "--prompt", "p",
+         "--registry-dir", str(reg_dir)],
+        capture_output=True, text=True, env=env, timeout=60,
+    ).stdout.strip()
+
+    record = json.loads((reg_dir / (job_id + ".json")).read_text(encoding="utf-8"))
+    record["broker"] = {"host": "127.0.0.1", "port": 65432, "tls": False}
+    (reg_dir / (job_id + ".json")).write_text(json.dumps(record), encoding="utf-8")
+
+    if entry == "cli":
+        cmd = [sys.executable, "-m", "lib_py.delegation.cli", "publish"]
+    else:
+        cmd = [sys.executable, str(SHIM_DIR / "publish_event.py")]
+    res = subprocess.run(
+        cmd + ["--job", job_id, "--event", "completed", "--detail", "done",
+               "--attempts", "1", "--registry-dir", str(reg_dir)],
+        capture_output=True, text=True, env=env, cwd=str(mam_sandbox), timeout=120,
+    )
+    assert res.returncode == 2, (res.stdout, res.stderr)
+    assert res.stdout == "", "publish must stay silent on stdout"
+    after = json.loads((reg_dir / (job_id + ".json")).read_text(encoding="utf-8"))
+    assert after["status"] == "completed", "status syncs even when the publish fails"
+    assert after.get("last_seq") == 1, "seq is consumed on failure by design"
+    audit = (Path(mam_sandbox) / ".mam" / "delegate_job_logs" / job_id / "events.ndjson")
+    events = [json.loads(l) for l in audit.read_text(encoding="utf-8").splitlines() if l.strip()]
+    published = [e for e in events if e.get("event") == "published"]
+    assert published and published[-1]["published"] is False
+    assert published[-1]["publish_error"]