自分で仕事を見つける agent

「呼ばれる」から「自分で探す」へ

s09 / s10 の teammate は受動的だ：メールボックスに誰かが message を送って初めて動き始め、完了したら idle に戻り、また呼ばれるまで待つ。

s11 の autonomous teammate は能動的だ：仕事がないときも本当に止まることなく、5秒ごとに .tasks/ ディレクトリをスキャンし、未認領のタスクを見つければ claim_task で認領して、作業を開始する。

Teammate lifecycle:
  +-------+
  | spawn |
  +---+---+
      |
      v
  +-------+
  | WORK  | <-- LLM tool_use ループ
  +---+---+
      |
      | stop_reason != tool_use OR tool_use("idle")
      v
  +-------+
  | IDLE  | 5秒ごとにポーリング、合計12回 = 60秒でタイムアウト
  +---+---+
      |
      +--> inbox に新しいメッセージ？  → WORK に復帰
      |
      +--> .tasks/ をスキャン、未認領タスク発見 → claim → WORK に復帰
      |
      +--> 60秒待っても何もなければ → status = shutdown、スレッド終了

Auto-claim · タスクボードから「仕事を取る」

核心関数：

def scan_unclaimed_tasks() -> list:
    unclaimed = []
    for f in sorted(TASKS_DIR.glob("task_*.json")):
        task = json.loads(f.read_text())
        if (task.get("status") == "pending"
                and not task.get("owner")
                and not task.get("blockedBy")):
            unclaimed.append(task)
    return unclaimed

3つのフィルタ条件 = pending + owner なし + blockedBy なし。一つでも欠けると不可。

claim 操作自体にはロック（_claim_lock）がある。2つの agent が同時に同じ未認領タスクを見て取り合うのを防ぐ——ロック内でもう一度読み直し、owner がまだ空であることを確認してから、「in_progress」とマークして owner をディスクに書き込む。

複数の agent が同時にタスクボードをスキャンするシミュレーション

3人の teammate が idle 状態で、3つのタスクが認領待ち。「次のスキャンラウンド」をクリックして、誰が最初に unclaimed タスクを取るかを確認しよう。

Identity 再注入 · 圧縮後も自分が誰か知っている

s06 で説明した auto_compact が messages を summary に置き換える。autonomous agent は1〜2時間動き続けると簡単に発火する——圧縮後、モデルは自分の名前も役割も知らない。

s11 の対策：新しいタスクを claim したとき、len(messages) ≤ 3 なら（直前に圧縮されたことを示す）、先頭に identity ブロックを自動挿入する：

if len(messages) <= 3:
    messages.insert(0, {
        "role": "user",
        "content": f"<identity>You are 'alice', role: coder, team: my-team. Continue your work.</identity>",
    })
    messages.insert(1, {"role":"assistant", "content": f"I am alice. Continuing."})

このパターンを identity reinjection と呼ぶ——人工的に構築した一往復の会話で「モデルが自分は誰か」を伝え、役割感覚を取り戻させる。