feat(agentctl): add foreground daemon coordinator with manager.lock

.doc/ops/CURRENT.md

@@ -18,6 +18,7 @@
 
 | owner | task | branch | worktree | mutex | status |
 |---|---|---|---|---|---|
+| Cursor | agentctl-daemon-2026-09-23 | feature/agentctl-daemon-2026-09-23 | .worktrees/agentctl-daemon-2026-09-23 | agentctl-daemon | in_progress |
 | Cursor | agentctl-handoff-2026-09-23 | feature/agentctl-handoff-2026-09-23 | .worktrees/agentctl-handoff-2026-09-23 | agentctl-handoff | in_progress |
 | Cursor | agentctl-claim-2026-09-23 | feature/agentctl-claim-2026-09-23 | .worktrees/agentctl-claim-2026-09-23 | agentctl-claim | done |
 | Cursor | agent-governance-config-2026-09-23 | feature/agent-governance-config-2026-09-23 | .worktrees/agent-governance-config-2026-09-23 | agent-governance-config | done |

.doc/ops/short-term/2026-09-23-agentctl-daemon.md

@@ -0,0 +1,47 @@
+---
+title: "W2-U4：常驻协调入口 daemon"
+task_id: agentctl-daemon-2026-09-23
+status: in_progress
+created: 2026-09-23
+updated: 2026-09-23
+owner: Cursor
+task_type: implementation
+task_tier: single
+policy_revision: "agent-governance-2026-09-21-r1 @ 981021b2b6e9081e9523f992634d273ef4631598"
+branch: feature/agentctl-daemon-2026-09-23
+worktree: .worktrees/agentctl-daemon-2026-09-23
+base_dev_sha: "981021b2b6e9081e9523f992634d273ef4631598"
+write_scope: ["scripts", "configs", ".doc"]
+allowed_environments: []
+mutex: agentctl-daemon
+review_state: pending
+integration_state: pending
+deployment: not_required
+cleanup: pending
+cleanup_owner: "独立维护者"
+cleanup_trigger: "合入 local_dev 且无进程占用后复查"
+remote_sync: not_requested
+---
+
+# W2-U4：常驻协调入口 daemon
+
+## 用户目标与授权
+
+- 用户目标：前台 `agentctl daemon run|status`，`manager.lock` 单实例，复用 `queue_schedule`；模板入库不装载。
+- 授权：W2-U4 实施；不 push；不装载 launchd/systemd。
+- 禁止：fork/setsid daemonize；把模板放入 `deploy/`；扩大权限。
+
+## 范围
+
+- `manager.lock` 独占 flock（不进 `physical_names`）
+- `daemon run --interval --max-cycles` 前台循环 + SIGTERM/SIGINT 优雅停
+- `daemon status` 只读
+- `scripts/daemon/` 下 systemd/launchd 模板 + README（明示未装载）
+
+## 验收
+
+1. 双实例并发 → 恰一存活，另一 `LOCK_BUSY` 带 `holder_pid`
+2. daemon 停止后 pending 候选不被自动合入
+3. 前台 CLI 语义不变
+4. daemon 无 register 越权
+5. unittest discover 全绿

scripts/agentctl.py

@@ -2126,6 +2126,189 @@ def claim_cancel(repo, args):
     return 0
 
 
+
+MANAGER_LOCK_NAME = 'manager.lock'
+DAEMON_HEARTBEAT_NAME = 'daemon.heartbeat'
+
+
+def _read_manager_holder_pid(path):
+    try:
+        raw = path.read_text().strip()
+    except OSError:
+        return None
+    if not raw:
+        return None
+    try:
+        return int(raw.split()[0])
+    except ValueError:
+        return None
+
+
+@contextlib.contextmanager
+def acquire_manager_lock(repo):
+    """Exclusive flock on manager.lock; not part of physical_names() lease set."""
+    path = repo.os_file(MANAGER_LOCK_NAME)
+    handle = path.open('a+')
+    try:
+        try:
+            fcntl.flock(handle, fcntl.LOCK_EX | fcntl.LOCK_NB)
+        except BlockingIOError:
+            holder_pid = _read_manager_holder_pid(path)
+            deny('LOCK_BUSY', 'Daemon manager lock is held',
+                 holder_pid=holder_pid, retry_after=1)
+        handle.seek(0)
+        handle.truncate()
+        handle.write(f'{os.getpid()}\n')
+        handle.flush()
+        os.fsync(handle.fileno())
+        yield handle
+    finally:
+        try:
+            fcntl.flock(handle, fcntl.LOCK_UN)
+        except OSError:
+            pass
+        handle.close()
+
+
+def write_daemon_heartbeat(repo, cycle):
+    payload = {
+        'pid': os.getpid(),
+        'updated': time.time(),
+        'cycle': cycle,
+    }
+    atomic_write(repo.os_file(DAEMON_HEARTBEAT_NAME), json.dumps(payload, sort_keys=True))
+
+
+def read_daemon_heartbeat(repo):
+    path = repo.root / 'os' / DAEMON_HEARTBEAT_NAME
+    if not path.exists() or path.is_symlink():
+        return None
+    try:
+        data = json.loads(path.read_text())
+    except (OSError, json.JSONDecodeError, UnicodeDecodeError):
+        return None
+    if not isinstance(data, dict):
+        return None
+    return data
+
+
+def manager_lock_status(repo):
+    """Read-only probe: never create files or take EX locks."""
+    path = repo.root / 'os' / MANAGER_LOCK_NAME
+    if not path.exists() or path.is_symlink():
+        return {'held': False, 'holder_pid': None, 'path': str(path)}
+    try:
+        with path.open('r') as handle:
+            try:
+                fcntl.flock(handle, fcntl.LOCK_SH | fcntl.LOCK_NB)
+            except BlockingIOError:
+                return {
+                    'held': True,
+                    'holder_pid': _read_manager_holder_pid(path),
+                    'path': str(path),
+                }
+            fcntl.flock(handle, fcntl.LOCK_UN)
+    except OSError as exc:
+        return {'held': False, 'holder_pid': None, 'path': str(path), 'error': str(exc)}
+    return {'held': False, 'holder_pid': None, 'path': str(path)}
+
+
+def daemon_run(repo, args):
+    """Foreground coordinator loop: exclusive manager.lock + queue_schedule cycles."""
+    interval = float(getattr(args, 'interval', 1.0) or 1.0)
+    if interval < 0:
+        deny('INVALID_REQUEST', 'daemon interval must be >= 0')
+    max_cycles = getattr(args, 'max_cycles', None)
+    if max_cycles is not None:
+        max_cycles = int(max_cycles)
+        if max_cycles < 1:
+            deny('INVALID_REQUEST', 'daemon max-cycles must be >= 1')
+    stop = {'flag': False}
+
+    def _request_stop(signum, frame):
+        stop['flag'] = True
+
+    previous_term = signal.signal(signal.SIGTERM, _request_stop)
+    previous_int = signal.signal(signal.SIGINT, _request_stop)
+    cycles = 0
+    try:
+        with acquire_manager_lock(repo):
+            emit({
+                'status': 'daemon_started',
+                'pid': os.getpid(),
+                'interval': interval,
+                'max_cycles': max_cycles,
+            })
+            while not stop['flag']:
+                cycles += 1
+                write_daemon_heartbeat(repo, cycles)
+                try:
+                    queue_schedule(repo, argparse.Namespace(drain=False))
+                except Denied as exc:
+                    detail = {
+                        'event': 'daemon_schedule_denied',
+                        'code': exc.code,
+                        'message': exc.message,
+                        **exc.details,
+                    }
+                    if exc.code != 'LOCK_BUSY':
+                        with repo.db(True) as db:
+                            ensure_schema(db)
+                            event(db, 'daemon', detail)
+                    emit({'status': 'denied', 'code': exc.code, 'message': exc.message,
+                          **exc.details}, sys.stderr)
+                if max_cycles is not None and cycles >= max_cycles:
+                    break
+                if stop['flag']:
+                    break
+                deadline = time.monotonic() + interval
+                while not stop['flag'] and time.monotonic() < deadline:
+                    time.sleep(min(0.05, max(0.0, deadline - time.monotonic())))
+            write_daemon_heartbeat(repo, cycles)
+            emit({'status': 'daemon_stopped', 'pid': os.getpid(), 'cycles': cycles,
+                  'reason': 'signal' if stop['flag'] else 'max_cycles'})
+            return 0
+    finally:
+        signal.signal(signal.SIGTERM, previous_term)
+        signal.signal(signal.SIGINT, previous_int)
+
+
+def daemon_status(repo, args):
+    """Read-only daemon probe: lock holder, heartbeat age, queue counts."""
+    try:
+        identity = repo.identity()
+    except Denied as exc:
+        emit({'status': 'denied', 'code': exc.code, 'message': exc.message, **exc.details})
+        return 1
+    lock = manager_lock_status(repo)
+    heartbeat = read_daemon_heartbeat(repo)
+    heartbeat_age = None
+    if heartbeat and isinstance(heartbeat.get('updated'), (int, float)):
+        heartbeat_age = max(0.0, time.time() - float(heartbeat['updated']))
+    pending = claimed = 0
+    try:
+        with repo.db() as db:
+            try:
+                pending = db.execute(
+                    "SELECT count(*) FROM queue_items WHERE state='pending'").fetchone()[0]
+                claimed = db.execute(
+                    "SELECT count(*) FROM queue_items WHERE state='claimed'").fetchone()[0]
+            except sqlite3.Error:
+                pending = claimed = 0
+    except Denied:
+        pass
+    emit({
+        'status': 'ok',
+        'lock_root': str(repo.root),
+        'epoch': identity.get('epoch'),
+        'manager_lock': lock,
+        'heartbeat': heartbeat,
+        'heartbeat_age_seconds': heartbeat_age,
+        'queue': {'pending': pending, 'claimed': claimed},
+    })
+    return 0
+
+
 def parser():
     p=argparse.ArgumentParser(description=__doc__);p.add_argument('--repo',default='.')
     sub=p.add_subparsers(dest='action',required=True)
@@ -2184,6 +2367,12 @@ def parser():
     qtn.add_parser('list')
     qrel=qtn.add_parser('release');qrel.add_argument('--run',required=True);qrel.add_argument('--reason',required=True)
     qrel.add_argument('--verified-by',required=True,dest='verified_by')
+    dm=sub.add_parser('daemon').add_subparsers(dest='daemon_action',required=True)
+    drun=dm.add_parser('run')
+    drun.add_argument('--interval',type=float,default=1.0,help='Seconds between schedule cycles')
+    drun.add_argument('--max-cycles',type=int,dest='max_cycles',
+                      help='Stop after N cycles (tests / bounded runs)')
+    dm.add_parser('status')
     return p
 
 
@@ -2192,7 +2381,7 @@ def main():
     write_actions = {
         'init', 'task', 'with', 'docs', 'validate', 'integrate', 'queue',
         'deploy', 'recover', 'freeze', 'unfreeze', 'hooks', 'guard', 'cleanup', 'claim',
-        'quarantine',
+        'quarantine', 'daemon',
     }
     # status/doctor remain readable without a policy file and never create one.
     if args.action in write_actions:
@@ -2204,6 +2393,8 @@ def main():
             require_policy(repo)  # status/list need schema constants but are non-mutating
         elif args.action == 'quarantine' and getattr(args, 'quarantine_action', None) == 'list':
             require_policy(repo)  # list needs TTL/schema constants
+        elif args.action == 'daemon' and getattr(args, 'daemon_action', None) == 'status':
+            pass  # read-only probe; never creates manager.lock or heartbeat
         elif args.action == 'cleanup':
             require_policy(repo)
         else:
@@ -2315,6 +2506,9 @@ def main():
         if args.quarantine_action=='sweep':return quarantine_sweep(repo)
         if args.quarantine_action=='list':return quarantine_list(repo)
         if args.quarantine_action=='release':return quarantine_release(repo,args)
+    elif args.action=='daemon':
+        if args.daemon_action=='run':return daemon_run(repo,args)
+        if args.daemon_action=='status':return daemon_status(repo,args)
     return 0
 
 

scripts/daemon/README.md

@@ -0,0 +1,14 @@
+# agentctl coordinator unit templates
+
+Templates only. **Not loaded** on this machine.
+
+- `agentctl-coordinator.service` — systemd `Type=simple` foreground unit
+- `com.agentctl.coordinator.plist` — launchd `KeepAlive` foreground job
+
+Loading requires **explicit authorization** plus a fail-closed verification record
+(owner, host, command evidence, rollback). Do not `launchctl load` / `systemctl enable`
+from a task worktree without that gate.
+
+Daemon runs as the caller; it only drives already-validated integrate candidates via
+`queue schedule` under exclusive `manager.lock`. It does not touch `env:*` and does
+not grant new capabilities.

scripts/daemon/agentctl-coordinator.service

@@ -0,0 +1,19 @@
+[Unit]
+Description=agentctl foreground coordinator (queue schedule under manager.lock)
+Documentation=file://%h/Agent-System/scripts/daemon/README.md
+After=network-online.target
+Wants=network-online.target
+
+[Service]
+Type=simple
+# WorkingDirectory must be the main worktree checked out on branch `dev`.
+WorkingDirectory=/path/to/Agent-System
+ExecStart=/usr/bin/python3 /path/to/Agent-System/scripts/agentctl.py daemon run --interval 1.0
+Restart=on-failure
+RestartSec=5
+# Runs as the unit user; no CapabilityBoundingSet escalation.
+KillSignal=SIGTERM
+TimeoutStopSec=30
+
+[Install]
+WantedBy=default.target

scripts/daemon/com.agentctl.coordinator.plist

@@ -0,0 +1,29 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!DOCTYPE plist PUBLIC "-//Apple//DTD PLIST 1.0//EN" "http://www.apple.com/DTDs/PropertyList-1.0.dtd">
+<plist version="1.0">
+<dict>
+	<key>Label</key>
+	<string>com.agentctl.coordinator</string>
+	<key>ProgramArguments</key>
+	<array>
+		<string>/usr/bin/python3</string>
+		<string>/path/to/Agent-System/scripts/agentctl.py</string>
+		<string>daemon</string>
+		<string>run</string>
+		<string>--interval</string>
+		<string>1.0</string>
+	</array>
+	<!-- WorkingDirectory must be the main worktree checked out on branch `dev`. -->
+	<key>WorkingDirectory</key>
+	<string>/path/to/Agent-System</string>
+	<key>RunAtLoad</key>
+	<true/>
+	<key>KeepAlive</key>
+	<true/>
+	<!-- Foreground process; launchd supervises without setsid daemonization. -->
+	<key>StandardOutPath</key>
+	<string>/tmp/agentctl-coordinator.out.log</string>
+	<key>StandardErrorPath</key>
+	<string>/tmp/agentctl-coordinator.err.log</string>
+</dict>
+</plist>

scripts/tests/test_agentctl.py

@@ -1049,6 +1049,62 @@ class AgentctlTests(unittest.TestCase):
         conn.close()
         self.assertEqual(state, "recovery_required")
 
+    def test_daemon_single_instance_lock_busy_with_holder_pid(self):
+        self.cli("init")
+        import concurrent.futures
+        cmd = [sys.executable, str(self.root / "scripts/agentctl.py"),
+               "daemon", "run", "--interval", "0", "--max-cycles", "3"]
+
+        def attempt():
+            return subprocess.run(cmd, cwd=self.root, text=True, capture_output=True, timeout=20)
+
+        with concurrent.futures.ThreadPoolExecutor(max_workers=2) as pool:
+            first = pool.submit(attempt)
+            second = pool.submit(attempt)
+            results = [first.result(), second.result()]
+        codes = sorted(r.returncode for r in results)
+        self.assertEqual(codes, [0, 75], [(r.returncode, r.stdout, r.stderr) for r in results])
+        busy = next(r for r in results if r.returncode == 75)
+        payload = json.loads(busy.stderr)
+        self.assertEqual(payload["code"], "LOCK_BUSY")
+        self.assertIn("holder_pid", payload)
+        self.assertIsInstance(payload["holder_pid"], int)
+
+    def test_daemon_stop_leaves_pending_queue_untouched(self):
+        task = self.initialized_task(scopes=(".",), capabilities=("git:dev",))
+        tip_before = self.git("rev-parse", "dev")
+        # Seed a pending item that is NOT validated — daemon must not promote it.
+        import sqlite3
+        import time as time_mod
+        import uuid
+        item_id = str(uuid.uuid4())
+        now = time_mod.time()
+        with sqlite3.connect(self.root / ".locks" / "state.sqlite") as db:
+            db.execute(
+                "INSERT INTO queue_items(id,task,head_sha,expected_dev,priority,state,created,updated)"
+                " VALUES(?,?,?,?,?,?,?,?)",
+                (item_id, "alpha", tip_before, tip_before, 0, "pending", now, now))
+            db.commit()
+        self.cli("daemon", "run", "--interval", "0", "--max-cycles", "2", cwd=self.root)
+        self.assertEqual(self.git("rev-parse", "dev"), tip_before)
+        with sqlite3.connect(self.root / ".locks" / "state.sqlite") as db:
+            state = db.execute("SELECT state FROM queue_items WHERE id=?", (item_id,)).fetchone()[0]
+        # Failed admit is ok; must not become completed/merged into tip.
+        self.assertNotEqual(state, "completed")
+        self.assertEqual(self.git("rev-parse", "dev"), tip_before)
+
+    def test_daemon_does_not_grant_register_or_docs_apply(self):
+        self.cli("init")
+        self.cli("daemon", "run", "--interval", "0", "--max-cycles", "1", cwd=self.root)
+        denied = self.cli("task", "register", "--id", "rogue", "--owner", "x", "--scope", "src",
+                          cwd=self.root, check=False)
+        self.assertNotEqual(denied.returncode, 0)
+        self.assertIn("WORKTREE_REQUIRED", denied.stderr)
+        status = json.loads(self.cli("daemon", "status", cwd=self.root).stdout)
+        self.assertEqual(status["status"], "ok")
+        self.assertIn("manager_lock", status)
+        self.assertIn("queue", status)
+
 
 class ProcessPlatformTests(unittest.TestCase):
     @classmethod