diff --git a/.doc/ops/CURRENT.md b/.doc/ops/CURRENT.md index 3abd1fef..4b7e2bd0 100644 --- a/.doc/ops/CURRENT.md +++ b/.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 | diff --git a/.doc/ops/short-term/2026-09-23-agentctl-daemon.md b/.doc/ops/short-term/2026-09-23-agentctl-daemon.md new file mode 100644 index 00000000..bc294f04 --- /dev/null +++ b/.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 全绿 diff --git a/scripts/agentctl.py b/scripts/agentctl.py index c3b2b6ff..57d69682 100755 --- a/scripts/agentctl.py +++ b/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 diff --git a/scripts/daemon/README.md b/scripts/daemon/README.md new file mode 100644 index 00000000..a7c2afb8 --- /dev/null +++ b/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. diff --git a/scripts/daemon/agentctl-coordinator.service b/scripts/daemon/agentctl-coordinator.service new file mode 100644 index 00000000..6afb3a1e --- /dev/null +++ b/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 diff --git a/scripts/daemon/com.agentctl.coordinator.plist b/scripts/daemon/com.agentctl.coordinator.plist new file mode 100644 index 00000000..f8701e24 --- /dev/null +++ b/scripts/daemon/com.agentctl.coordinator.plist @@ -0,0 +1,29 @@ + + + + + Label + com.agentctl.coordinator + ProgramArguments + + /usr/bin/python3 + /path/to/Agent-System/scripts/agentctl.py + daemon + run + --interval + 1.0 + + + WorkingDirectory + /path/to/Agent-System + RunAtLoad + + KeepAlive + + + StandardOutPath + /tmp/agentctl-coordinator.out.log + StandardErrorPath + /tmp/agentctl-coordinator.err.log + + diff --git a/scripts/tests/test_agentctl.py b/scripts/tests/test_agentctl.py index f62dc77b..b9d43dab 100644 --- a/scripts/tests/test_agentctl.py +++ b/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