Day 3｜最小可行常駐：讓 agent 自己醒來

我的排程有一次全滅了三天，沒有任何一個任務跑起來，也沒有任何一行錯誤日誌。

事後查出兩個原因疊在一起：一是事件匯流排上沒有任何訂閱者，事件送出去就掉進虛空；二是鎖檔的路徑跟著工作目錄漂移，兩個行程各自拿到「自己的」鎖，於是誰也沒擋住誰，但真正該跑的那個分支被跳過了。

兩個都是「觸發層」的問題。今天把觸發層講完，這是整個 harness 裡最不性感、但壞掉最難察覺的一層。

三種醒來的方式

多數人從第一種開始，這是對的。但如果你只有第一種，你的 agent 就只會在整點上工，其他時間發生的事全部要等到下一個整點。

三種觸發源最後都收斂成同一個佇列，這樣併發控制、互斥、執行紀錄才有單一的地方可以做。

我自己的做法是一個常駐程序同時餵這三種：內部有 tick，外部有 webhook 進來，另外有幾條長連線在監看資料。它們全部收斂成同一個「有事要處理」的佇列。

為什麼我最後沒用 cron

一開始我用系統 cron，跑了大概兩週就換掉。三個原因：

環境不對。 cron 的 PATH 只有很基本的幾個目錄。你在終端機打 claude 找得到，cron 裡面 command not found。這件事我後來寫了一個探測函式解決，等一下貼出來。

看不到狀態。 cron 只告訴你「有沒有執行」，不告訴你「執行到哪」。任務跑了 20 分鐘還沒結束，你不知道它是在工作還是卡死。

沒有併發控制。 兩個任務同時到期，cron 就開兩個。開到第五個的時候你的機器風扇開始叫。

換成常駐程序以後，這三件事都在同一個地方處理。代價是你要自己管這個程序的生命週期（開機自動啟動、掛掉自動重啟），在 macOS 是 launchd，Linux 是 systemd。

CLI 到底裝在哪

這是每個要在背景叫 CLI 的人都會踩的坑。不同的安裝方式會把 binary 放在完全不同的地方：

import os, pathlib, shutil

def which_claude() -> str | None:
    """依序探測常見安裝位置，回傳第一個存在且可執行的路徑。"""
    # 1. PATH 裡找得到就直接用
    if p := shutil.which("claude"):
        return p

    home = pathlib.Path.home()
    candidates = [
        "/opt/homebrew/bin/claude",          # Homebrew (Apple Silicon)
        "/usr/local/bin/claude",             # Homebrew (Intel) / 手動安裝
        home / ".bun/bin/claude",            # Bun
        home / ".volta/bin/claude",          # Volta
        home / ".claude/bin/claude",         # 官方安裝script
        home / ".local/bin/claude",          # pipx 風格
        home / ".asdf/shims/claude",         # asdf
    ]
    # NVM 會把每個 Node 版本各裝一份，要掃版本目錄
    nvm = home / ".nvm/versions/node"
    if nvm.is_dir():
        candidates += [d / "bin/claude" for d in sorted(nvm.iterdir(), reverse=True)]

    for c in candidates:
        c = pathlib.Path(c)
        if c.is_file() and os.access(c, os.X_OK):
            return str(c)
    return None

這段程式我在自己的平台裡改過四次，每次都是因為某個使用者的環境長得跟我想的不一樣。如果你的 agent 要給別人用，這種探測邏輯遲早要寫。

一個小提醒：探測結果要快取，但不要永久快取。使用者升級 CLI、換安裝方式的時候你要重新探。我的做法是快取到程序重啟為止。

最小的常駐排程器

下面這個 120 行的東西，取代了我原本的 cron。它處理定時、互斥、併發上限、以及最重要的「執行紀錄」。

#!/usr/bin/env python3
"""最小常駐排程器：定時喚醒 agent，帶併發上限與執行紀錄。"""
import asyncio, json, pathlib, subprocess, time, uuid
from dataclasses import dataclass, asdict
from datetime import datetime, timezone

HOME = pathlib.Path.home() / ".myagent"
HOME.mkdir(exist_ok=True)
RUNS = HOME / "runs.jsonl"          # 執行紀錄，一行一筆

MAX_CONCURRENT = 2                   # 同時最多跑幾個
sem = asyncio.Semaphore(MAX_CONCURRENT)

@dataclass
class Task:
    name: str
    interval_secs: int
    prompt: str
    last_run: float = 0.0

def record(entry: dict) -> None:
    """執行紀錄一定要落地。沒有紀錄的自動化等於沒有自動化。"""
    entry["ts"] = datetime.now(timezone.utc).isoformat()
    with RUNS.open("a", encoding="utf-8") as f:
        f.write(json.dumps(entry, ensure_ascii=False) + "\n")

async def run_task(task: Task, claude: str) -> None:
    run_id = uuid.uuid4().hex[:8]
    async with sem:                              # 併發上限
        record({"run_id": run_id, "task": task.name, "event": "start"})
        started = time.time()
        try:
            proc = await asyncio.create_subprocess_exec(
                claude, "-p", task.prompt,
                "--allowedTools", "Read,Grep,Glob",
                stdout=asyncio.subprocess.PIPE,
                stderr=asyncio.subprocess.PIPE,
            )
            out, err = await asyncio.wait_for(proc.communicate(), timeout=900)
            record({
                "run_id": run_id, "task": task.name, "event": "finish",
                "exit_code": proc.returncode,
                "elapsed": round(time.time() - started, 1),
                "output_head": out.decode(errors="replace")[:300],
                "stderr_tail": err.decode(errors="replace")[-300:],
            })
        except asyncio.TimeoutError:
            proc.kill()
            record({"run_id": run_id, "task": task.name, "event": "timeout"})
        except Exception as e:
            record({"run_id": run_id, "task": task.name,
                    "event": "error", "detail": repr(e)[:300]})

async def main() -> None:
    claude = which_claude()
    if not claude:
        raise SystemExit("找不到 claude CLI，請先安裝或設定路徑")

    tasks = [
        Task("錯誤日誌巡檢", 3600, "檢查最近一小時的錯誤日誌，有異常摘要成三行"),
        Task("每日報告", 86400, "彙整今天的工作紀錄，產出一份日報"),
    ]

    running: set[asyncio.Task] = set()
    while True:
        now = time.time()
        for t in tasks:
            if now - t.last_run >= t.interval_secs:
                t.last_run = now
                job = asyncio.create_task(run_task(t, claude))
                running.add(job)
                job.add_done_callback(running.discard)
        await asyncio.sleep(30)      # tick 每 30 秒一次

if __name__ == "__main__":
    asyncio.run(main())

三個設計決定值得說明。

tick 30 秒，不是每秒。 排程精度到分鐘就夠了，30 秒的 tick 讓 CPU 幾乎不動。

Semaphore 而不是佇列。 超過上限的任務會等，不會被丟掉。如果你希望超過上限就丟棄（例如高頻的監控），把 async with sem 換成 sem.locked() 判斷後直接 return，並且記一筆 event: skipped。丟掉沒關係，丟掉而不留紀錄才是問題。

每一筆都寫 runs.jsonl。 start、finish、timeout、error 四種事件都寫。我前面講的那次三天全滅之所以難查，就是因為當時沒有這個檔案，我只能從結果反推，而結果是「什麼都沒發生」。

併發之外還要防連環爆

上面的 semaphore 只防同時間的併發。還有一種情況它防不住：agent A 完成後觸發 agent B，B 又觸發 C，一路傳下去。

我後來加了兩道保險：

一是跳數上限。每次委派都把 hop 深度加一，寫進環境變數傳給子行程，超過 5 就拒絕執行。

hop = int(os.environ.get("AGENT_HOP_DEPTH", "0"))
if hop >= 5:
    record({"event": "hop_limit_exceeded", "hop": hop})
    raise SystemExit(0)
env = {**os.environ, "AGENT_HOP_DEPTH": str(hop + 1)}

二是滑動視窗斷路器。統計最近 N 分鐘內的 spawn 次數，超過閾值就整個停下來並通知我。這個機制在我的平台裡是跨行程的，狀態寫在一個帶檔案鎖的 JSON 裡，因為 spawn 可能來自好幾個不同的進入點。

代價

常駐程序的第一個代價是，它自己可能會死。而它死掉的時候，沒有任何 agent 會醒來告訴你這件事，因為叫醒它們的就是這個死掉的程序。

我的處理方式是讓它定期寫一個 heartbeat 檔案，然後用系統層級的 launchd/systemd 監看：程序沒了就重啟。監看的東西必須比被監看的東西更簡單、更可靠，這是唯一一條可以依賴的規則。

第二個代價是狀態全部在記憶體。程序重啟以後 last_run 歸零，所有任務會立刻各跑一次。如果你的任務是「發日報」，使用者就會在重啟的瞬間收到一份莫名其妙的日報。把 last_run 一起寫進磁碟可以解決，代價是多一次 I/O。

明天處理更難的一件事：它醒過來了，但它不知道自己昨天做過什麼。

那篇會講 session 和 conversation 的差別，以及為什麼 CLI 的 --resume 在多對話場景下會出事。