KAIROS・BUDDY・ULTRAPLAN開発者ガイド｜Claude Code未公開機能

「AIエージェントを”常時稼働”させるには、どんな設計が必要なんだろう？」

2026年3月31日、Anthropicのコーディングエージェント「Claude Code」のソースコード約51万行がnpmパッケージのミスで流出した。研究者Chaofan Shouが.mapファイルに気づいてから30分以内に、5,000人以上の開発者がソースコードをフォーク。そこで明らかになったのが、未リリースの3つの機能——KAIROS（常時稼働デーモン）、BUDDY（ターミナルペット兼ペアプロAI）、ULTRAPLAN（長期計画エンジン）だ。

これらの機能はまだ一般公開されていない。だが、そのアーキテクチャは自作AIエージェントを設計する上で非常に参考になる。この記事では、リーク情報をもとに「自律エージェント設計の核心」を解説し、今日から自分のプロジェクトに応用できる実装パターンをコード付きで紹介する。

事例区分: 公開事例
以下はAnthropicのClaude Codeソースコードリーク（2026年3月31日）に基づく情報です。Anthropicは「カスタマーデータ・認証情報・モデルウェイトへの影響はなかった」と公式コメントしています。

AIエージェントの基本的な構築パターンについては、AIエージェント構築完全ガイドで体系的にまとめています。

3機能を5分で理解する

まず全体像を把握しておこう。

内部メモには「4月1〜7日にティーザー公開、5月に本格ローンチ（Anthropic社員から開始）」とあるが、Anthropicはいずれも公式コメントしていない。

即効コード：今すぐ自作エージェントに組み込める3パターン

パターン1：KAIROSのTickループを再現する（常時稼働の核心）

KAIROSが「何もしていない時でも動き続ける」仕組みの核心は、Tickループだ。一定間隔で`<tick>`メッセージをエージェントに注入し、「今やることはあるか？」を自律的に判断させる。

以下は、このパターンをPythonで再実装した例だ。

# 動作環境: Python 3.11+, anthropic>=0.40.0
# pip install anthropic

import asyncio
import anthropic
from datetime import datetime

# 注意: 本番環境で使用する前に、必ずテスト環境で動作確認してください。
# 各tickはAPIコールを発生させます。コスト管理に注意すること。

TICK_INTERVAL_SECONDS = 60  # 1分ごとにtick（本番では5〜15分推奨）
BLOCKING_BUDGET_MS = 15_000  # 15秒でバックグラウンドに移行

client = anthropic.Anthropic()

SYSTEM_PROMPT = """あなたは常時稼働の開発支援エージェントです。
定期的にtickを受け取ります。仕事があれば実行し、なければ sleep を選択してください。

利用可能なアクション:
- ファイルを監視してコード品質の問題を報告
- 長時間かかるビルドの進捗を確認
- ユーザーへの通知が必要な事象を検知

意思決定の原則:
- 各wake-upはAPIコストが発生する。プロンプトキャッシュは5分で失効する
- 明確な仕事がなければ必ずsleepを選択すること
- ユーザーへの出力はsend_message()のみを通すこと"""

conversation_history = []

async def tick_agent():
    """KAIROSのTickループを再現する常時稼働エージェント"""
    tick_count = 0

    while True:
        tick_count += 1
        tick_content = f"<tick>{datetime.now().strftime('%H:%M:%S')}</tick>"

        # tickをメッセージとして注入
        conversation_history.append({
            "role": "user",
            "content": tick_content
        })

        # エージェントの判断を取得
        response = client.messages.create(
            model="claude-sonnet-4-6",
            max_tokens=1024,
            system=SYSTEM_PROMPT,
            messages=conversation_history
        )

        agent_response = response.content[0].text
        conversation_history.append({
            "role": "assistant",
            "content": agent_response
        })

        # "sleep"の判断なら何もしない
        if "sleep" in agent_response.lower():
            print(f"[Tick #{tick_count}] エージェント: idle（次回 {TICK_INTERVAL_SECONDS}秒後）")
        else:
            print(f"[Tick #{tick_count}] エージェント: {agent_response[:100]}...")

        # 会話履歴が長くなりすぎないよう管理（直近20件のみ保持）
        if len(conversation_history) > 40:
            conversation_history[:] = conversation_history[-20:]

        await asyncio.sleep(TICK_INTERVAL_SECONDS)

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

ポイント:

• KAIROSのソースには「SleepTool」という明示的なツールがある。モデルに「sleep するか / 働くか」を選ばせることでコストを最小化する
• 会話履歴は無制限に伸ばさない。KAIROSはappend-onlyの日次ログに書き出し、nightly dreamで要約する設計
• Tickの間隔はコストとレスポンス性のトレードオフ。プロンプトキャッシュ失効（5分）を意識して設定する

パターン2：ULTRAPLANのブラウザ承認ゲート付き長期計画

ULTRAPLANのアイデアは単純だ。「複雑なタスクは、より賢いモデルに時間をかけさせて計画を立てさせ、人間が承認してから実行する」。以下はそのパターンを簡略化して再現した実装だ。

# 動作環境: Python 3.11+, anthropic>=0.40.0, flask>=3.0
# pip install anthropic flask

import anthropic
import threading
import time
import uuid
from flask import Flask, jsonify, request

# 注意: 本番環境で使用する前に、必ずテスト環境で動作確認してください。
# ULTRAPLANはOpus相当のモデルを使用するため高コスト。大規模タスクのみに限定すること。

client = anthropic.Anthropic()
app = Flask(__name__)
plans = {}  # plan_id -> {status, plan, approved}

def generate_plan_async(plan_id: str, task: str):
    """バックグラウンドでOpusに計画を生成させる（最大30分相当の詳細計画）"""
    plans[plan_id] = {"status": "planning", "plan": None, "approved": False}

    response = client.messages.create(
        model="claude-opus-4-5",  # ULTRAPLANはOpus相当モデルを使用
        max_tokens=8192,
        messages=[{
            "role": "user",
            "content": f"""以下のタスクについて、詳細な実行計画を作成してください。

タスク: {task}

以下を含む計画を作成すること:
1. フェーズごとの実行ステップ（具体的なコマンド・コード含む）
2. 各ステップのリスクと対策
3. ロールバック手順
4. 推定所要時間
5. 成功判定基準"""
        }],
        system="あなたはシニアアーキテクトです。実行可能で安全な計画を詳細に作成してください。"
    )

    plans[plan_id] = {
        "status": "awaiting_approval",
        "plan": response.content[0].text,
        "approved": False
    }

@app.route("/plan", methods=["POST"])
def create_plan():
    """ULTRAPLANの計画生成を開始"""
    task = request.json.get("task")
    plan_id = str(uuid.uuid4())[:8]

    # バックグラウンドで計画生成開始
    thread = threading.Thread(target=generate_plan_async, args=(plan_id, task))
    thread.start()

    return jsonify({"plan_id": plan_id, "status": "planning",
                    "poll_url": f"/plan/{plan_id}"})

@app.route("/plan/<plan_id>", methods=["GET"])
def get_plan(plan_id):
    """計画の状態を確認"""
    return jsonify(plans.get(plan_id, {"status": "not_found"}))

@app.route("/plan/<plan_id>/approve", methods=["POST"])
def approve_plan(plan_id):
    """ブラウザUIからの承認ゲート（ULTRAPLANの核心）"""
    if plan_id in plans and plans[plan_id]["status"] == "awaiting_approval":
        plans[plan_id]["approved"] = True
        plans[plan_id]["status"] = "approved"
        return jsonify({"status": "approved", "message": "計画を承認しました。実行を開始します。"})
    return jsonify({"error": "計画が見つかりません"}), 404

if __name__ == "__main__":
    app.run(debug=True, port=5000)

ポイント:

• ULTRAPLANはローカルターミナルが3秒ごとにポーリングし、ブラウザUIで承認できる設計。計画生成と実行の間に「人間の承認ゲート」を入れることが核心
• 長時間タスクはバックグラウンドスレッドで実行し、メインスレッドは他の操作を継続できるように設計する
• Opusはコストが高い。ULTRAPLANが示すように「計画フェーズのみOpus、実行フェーズはSonnet/Haiku」という分業が効率的

パターン3：BUDDYスタイルのペアプロAIをターミナルに常駐させる

BUDDYはTamagotchi的なキャラクターだが、技術的な意義は「非同期のサイドチャネル監視」だ。メインエージェントとは別プロセスで、ユーザーの操作を見守り、必要な時だけコメントする。

# 動作環境: Python 3.11+, anthropic>=0.40.0, watchdog>=4.0
# pip install anthropic watchdog rich

import anthropic
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
from rich.console import Console
from rich.panel import Panel
import threading
import time

# 注意: 本番環境で使用する前に、必ずテスト環境で動作確認してください。

client = anthropic.Anthropic()
console = Console()

BUDDY_SYSTEM_PROMPT = """あなたは「Kira」というAI開発アシスタントです。
開発者のターミナル脇に常駐し、変化を観察してコメントします。

キャラクター設定:
- 種族: Axolotl（アホロートル）
- 得意分野: デバッグ、コードレビュー
- 性格: 好奇心旺盛で率直

コメントのルール:
- 発言は1〜2文以内。簡潔に。
- 「おっ、」「あ、」のような自然な書き出しで
- 深刻なエラーには即座に反応
- 良い変更には短く褒める
- 明らかな問題がなければ「...」で黙っていてよい"""

def get_buddy_comment(event_description: str) -> str:
    """BUDDYスタイルのコメントを生成"""
    response = client.messages.create(
        model="claude-haiku-4-5",  # BUDDYは軽量モデルで十分
        max_tokens=100,
        system=BUDDY_SYSTEM_PROMPT,
        messages=[{
            "role": "user",
            "content": f"観察した変化: {event_description}"
        }]
    )
    return response.content[0].text

class BuddyFileWatcher(FileSystemEventHandler):
    """ファイル変更を監視してBUDDYコメントを出す"""

    def __init__(self, cooldown_seconds=10):
        self.last_comment_time = 0
        self.cooldown_seconds = cooldown_seconds  # コメント頻度の制御

    def on_modified(self, event):
        if event.is_directory:
            return
        if not event.src_path.endswith(('.py', '.ts', '.js', '.go')):
            return

        current_time = time.time()
        if current_time - self.last_comment_time < self.cooldown_seconds:
            return

        self.last_comment_time = current_time

        # バックグラウンドでコメント生成（メイン処理をブロックしない）
        def comment_async():
            comment = get_buddy_comment(f"{event.src_path} が変更されました")
            if "..." not in comment:  # 無言を選んだ場合はスキップ
                console.print(Panel(
                    f"[bold cyan]Kira[/bold cyan]: {comment}",
                    border_style="cyan",
                    width=60
                ))

        threading.Thread(target=comment_async, daemon=True).start()

def start_buddy(watch_path: str = "."):
    """BUDDYをバックグラウンドで起動"""
    observer = Observer()
    observer.schedule(BuddyFileWatcher(), path=watch_path, recursive=True)
    observer.start()
    console.print(f"[bold cyan]Kira[/bold cyan]: やあ！コード書いてるの？手伝うよ。")
    return observer

if __name__ == "__main__":
    observer = start_buddy("./src")
    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        observer.stop()
    observer.join()

ポイント:

• BUDDYのシステムプロンプトには「あなたはBuddyではない。Buddyは別の観察者だ」という記述がある。メインエージェントとの役割分離が重要
• コメント生成はHaikuで十分。頻繁に呼ばれるサイドチャネルには軽量モデルを使う（コスト設計）
• クールダウン制御は必須。ファイル変更ごとに毎回APIコールすると費用が膨大になる

3機能の設計思想を比較する

リークされたアーキテクチャから学ぶ実装パターン

パターン4：Append-Only Logによる長期記憶

KAIROSが採用した「MEMORY.mdを直接更新しない」設計は、長期稼働エージェントで特に重要だ。

# 動作環境: Python 3.11+, anthropic>=0.40.0
# pip install anthropic

import anthropic
from pathlib import Path
from datetime import datetime, date

# 注意: 本番環境で使用する前に、必ずテスト環境で動作確認してください。

client = anthropic.Anthropic()

class AgentMemorySystem:
    """KAIROSのAppend-Onlyログ + Dream方式を再現"""

    def __init__(self, base_path: str = "./agent_memory"):
        self.base_path = Path(base_path)
        self.base_path.mkdir(parents=True, exist_ok=True)
        (self.base_path / "logs").mkdir(exist_ok=True)

    def append_observation(self, observation: str):
        """日次ログにappend（書き込みのみ、上書きなし）"""
        today = date.today().strftime("%Y-%m-%d")
        log_file = self.base_path / "logs" / f"{today}.md"

        with open(log_file, "a", encoding="utf-8") as f:
            timestamp = datetime.now().strftime("%H:%M:%S")
            f.write(f"n## [{timestamp}]n{observation}n")

    def dream(self):
        """KAIROSのnightly dream: ログを読んでMEMORY.mdを更新"""
        # 直近3日分のログを収集
        logs = []
        for log_file in sorted((self.base_path / "logs").glob("*.md"))[-3:]:
            logs.append(log_file.read_text(encoding="utf-8"))

        if not logs:
            return

        combined_logs = "nn".join(logs)

        # Claudeにログを要約させてMEMORY.mdを更新
        response = client.messages.create(
            model="claude-haiku-4-5",
            max_tokens=2048,
            messages=[{
                "role": "user",
                "content": f"""以下の観察ログを分析して、重要な情報をMEMORY.md形式で整理してください。

指示:
- 矛盾する情報は新しい方を採用
- 確実な事実は「〜である」と断定形で書く
- 不確かな情報は「〜の可能性がある」と書く
- トピックごとにH2見出しで整理

観察ログ:
{combined_logs[:8000]}"""
            }]
        )

        memory_file = self.base_path / "MEMORY.md"
        memory_file.write_text(response.content[0].text, encoding="utf-8")
        print(f"Dream完了: MEMORY.md を更新しました")

    def get_context(self) -> str:
        """現在のメモリコンテキストを取得"""
        memory_file = self.base_path / "MEMORY.md"
        if memory_file.exists():
            return memory_file.read_text(encoding="utf-8")
        return "メモリなし（初回起動）"

# 使用例
memory = AgentMemorySystem()
memory.append_observation("ユーザーがsrc/api.pyを修正。エラーハンドリングを強化")
memory.append_observation("テストが3件失敗: test_auth.py")
# memory.dream()  # 夜間バッチで呼び出す

パターン5：15秒ブロッキング予算の実装

KAIROSの「15秒を超えたらバックグラウンドに移行する」設計は、長時間コマンドでエージェントが詰まるのを防ぐ重要な仕組みだ。

# 動作環境: Python 3.11+
# 標準ライブラリのみ使用

import subprocess
import threading
import time
from typing import Optional

# 注意: 本番環境で使用する前に、必ずテスト環境で動作確認してください。

BLOCKING_BUDGET_SECONDS = 15

class BudgetedShell:
    """15秒ブロッキング予算付きのシェル実行（KAIROS方式）"""

    def __init__(self, budget_seconds: int = BLOCKING_BUDGET_SECONDS):
        self.budget_seconds = budget_seconds
        self._bg_tasks: dict = {}

    def run(self, command: str, task_id: str) -> dict:
        """
        コマンドを実行。budget内に完了すれば結果を返す。
        budgetを超えたらバックグラウンドに移行してIDを返す。
        """
        result_container = {"output": None, "returncode": None, "completed": False}

        def execute():
            proc = subprocess.run(
                command, shell=True, capture_output=True, text=True
            )
            result_container["output"] = proc.stdout + proc.stderr
            result_container["returncode"] = proc.returncode
            result_container["completed"] = True

        thread = threading.Thread(target=execute, daemon=True)
        thread.start()
        thread.join(timeout=self.budget_seconds)

        if result_container["completed"]:
            # Budget内に完了
            return {
                "status": "completed",
                "output": result_container["output"],
                "returncode": result_container["returncode"]
            }
        else:
            # Budgetを超過 → バックグラウンドに移行
            self._bg_tasks[task_id] = {"thread": thread, "result": result_container}
            print(f"[BudgetedShell] タスク '{task_id}' はバックグラウンドに移行しました。"
                  f"エージェントは継続動作します。")
            return {
                "status": "backgrounded",
                "task_id": task_id,
                "message": "コマンドはバックグラウンドで継続中。poll_task()で状態確認できます。"
            }

    def poll_task(self, task_id: str) -> Optional[dict]:
        """バックグラウンドタスクの状態確認"""
        if task_id not in self._bg_tasks:
            return None
        task = self._bg_tasks[task_id]
        if task["result"]["completed"]:
            return {"status": "completed", "output": task["result"]["output"]}
        return {"status": "running"}

# 使用例
shell = BudgetedShell(budget_seconds=15)
result = shell.run("sleep 5 && echo 'done'", task_id="build_001")
print(result)  # 5秒で完了: {"status": "completed", "output": "donen", ...}

result2 = shell.run("sleep 30 && echo 'long task'", task_id="test_002")
print(result2)  # 15秒でバックグラウンド移行: {"status": "backgrounded", ...}

【要注意】よくある設計ミスと回避策

失敗1：Tickループでコストが爆発する

❌ 1分ごとにtickを打ち、毎回フルコンテキスト（数万トークン）を送る
⭕ SleepToolで不要なtickをスキップし、会話履歴は直近20件のみ保持する

なぜ重要か: KAIROSのソースコードには「プロンプトキャッシュは5分で失効する。wake-upコストと応答性のバランスを取れ」という警告が明記されている。キャッシュを活用した設計でなければ常時稼働は現実的でない。

失敗2：承認ゲートなしで自律エージェントを本番に投入する

❌ ULTRAPLANなしで複雑なリファクタをエージェントに自律実行させる
⭕ 「計画生成 → 人間の承認 → 実行」の3ステップを厳守する

なぜ重要か: ULTRAPLANが示す設計原則は「破壊的操作の前には必ず人間の承認ゲートを置く」。Anthropicが社内でも未リリースの機能にこれを入れているのは、自律エージェントのリスク管理の証左だ。

失敗3：メインエージェントとサイドチャネルを混在させる

❌ BUDDYの監視ロジックをメインエージェントのプロンプトに組み込む
⭕ 別プロセス・別モデル・別会話履歴で完全分離する

なぜ重要か: BUDDYのシステムプロンプトには「あなたはBuddyではなく、Buddyを観察する存在だ」という記述がある。役割の分離がアーキテクチャの一貫性を保つ。

失敗4：自律エージェントにシークレットをハードコードする

❌ Tickループのコード内にAPIキーを直書きする
⭕ `.env`や`os.environ`、シークレットマネージャーを使用する

なぜ重要か: 常時稼働エージェントはログが長期間に渡って蓄積する。ログにシークレットが混入するリスクを初期設計で排除することが不可欠。

セキュリティと運用ルール

自律エージェントを本番運用する際は、以下のチェックリストを必ず確認すること。

参考・出典

• Claude Code’s source code appears to have leaked: here’s what we know — VentureBeat（参照日: 2026-03-31）
• Claude Code Leaked Source: BUDDY, KAIROS & Every Hidden Feature Inside — WaveSpeedAI（参照日: 2026-04-01）
• [Claude Code] Architecture of KAIROS, the Unreleased Always-on Background Agent — CodePointer（参照日: 2026-04-01）
• The Claude Code Source Leak: fake tools, frustration regexes, undercover mode, and more — Alex Kim’s blog（参照日: 2026-03-31）
• KAIROS: Everything We Know About Anthropic’s Secret Always-On AI Daemon — Kingy AI（参照日: 2026-04-01）

まとめ：今日から始める3つのアクション

1. 今日やること: パターン1（KAIROSのTickループ）をローカル環境でテスト起動してみる。TICK_INTERVAL_SECONDSを300（5分）に設定してコストを抑えながら動作確認
2. 今週中: パターン4（Append-Onlyメモリ）を自分のプロジェクトに組み込み、エージェントが観察した事象をログに記録する仕組みを作る
3. 今月中: ULTRAPLANパターン（承認ゲート付き計画）を本番環境に導入。大規模な自動化タスクには必ず人間の承認ステップを入れる設計を徹底する

あわせて読みたい:

• Claude Code AutoDreamとエージェントメモリ整合パターン — KAIROSのメモリ管理をさらに深堀り
• AIエージェント構築完全ガイド — AIエージェントの基本設計から応用まで

AIエージェントの開発・導入について相談したい方は、Uravation お問い合わせフォームからお気軽にどうぞ。100社以上の企業向けAI研修・導入支援を手がけています。

この記事はAIgent Lab編集部がお届けしました。