Chapter 08 — Production Daemon, Skills, and Plugins¶
Companion to book/ch08_*.md. Runs top-to-bottom in Google Colab in mock mode with no API key required.
import os
if not os.path.exists("crafting-agentic-swarms"):
!git clone https://github.com/TheAiSingularity/crafting-agentic-swarms.git
%cd crafting-agentic-swarms
!pip install -e ".[dev]" --quiet
!pip install matplotlib plotly ipywidgets --quiet
import os
try:
from google.colab import userdata
os.environ["ANTHROPIC_API_KEY"] = userdata.get("ANTHROPIC_API_KEY")
print("Using real API (key from Colab secrets).")
except (ImportError, Exception):
os.environ.setdefault("SWARM_MOCK", "true")
print("Running in mock mode (no API key needed).")
What you'll build here¶
- Run the KAIROS daemon with
execute_mode=Falseagainst 3 sample tasks. - Read the append-only tick log back and render it as a timeline.
- Build a small
SkillLibrary, add 3 skills, and search it by keyword similarity. - Inspect a Claude Code plugin directory structure (real if available, synthetic fallback).
1. Why a daemon¶
Agent sessions end when the user closes the tab. Real systems need a heartbeat — a process that wakes up every few minutes, checks pending work, and decides whether to act. KAIROS is the reference daemon: safe by default, append-only log, defers irreversible actions to a human.
2. Spin up KAIROS in a temp directory¶
We set execute_mode=False so the daemon only logs its decisions without launching the swarm. For a classroom notebook that keeps each tick deterministic and cheap.
import tempfile
from pathlib import Path
from swarm.daemon.kairos import KairosDaemon
log_dir = Path(tempfile.mkdtemp()) / "kairos"
daemon = KairosDaemon(log_dir=log_dir, tick_interval=1, execute_mode=False)
print(f"Log dir: {log_dir}")
3. Queue three tasks¶
add_task pushes onto the daemon's inbox. add_event pushes onto the external-event queue. On each tick the daemon inspects both plus the log tail and decides what to do next.
tasks = [
"Review open PRs and post a daily summary to Slack.",
"Archive stale issues older than 30 days.",
"Run the nightly SWE-bench smoke and report pass rate.",
]
for t in tasks:
daemon.add_task(t)
print(f"{len(daemon._pending_tasks)} tasks queued")
4. Tick manually three times¶
Instead of starting the loop, we call _tick() directly so each tick completes before the next starts. The log records every decision including parse errors from the mock LLM.
for _ in range(3):
await daemon._tick()
log_path = next(log_dir.glob("*.log"))
print(log_path.read_text())
5. Parse the log¶
Each line is [iso-timestamp] message. We parse it into a list of (datetime, str) tuples so we can plot the timeline. Simple regex suffices; never reach for a heavy parser until you need to.
import re
from datetime import datetime
entries = []
for line in log_path.read_text().splitlines():
m = re.match(r"\[(.*?)\] (.*)", line)
if not m:
continue
ts = datetime.fromisoformat(m.group(1))
entries.append((ts, m.group(2)))
for ts, msg in entries:
print(f"{ts.strftime('%H:%M:%S')} {msg[:80]}")
6. Render the tick log as a timeline¶
Scatter plot of tick events. For a real daemon you would pipe this into Grafana or a TUI, but matplotlib is fine for a notebook walkthrough.
import matplotlib.pyplot as plt
import matplotlib.dates as mdates
ts_list = [e[0] for e in entries]
fig, ax = plt.subplots(figsize=(10, 2.8))
ax.scatter(ts_list, [1] * len(ts_list), s=80)
for ts, msg in entries:
ax.annotate(msg[:35], (ts, 1), xytext=(0, 12), textcoords="offset points",
rotation=30, fontsize=7)
ax.yaxis.set_visible(False)
ax.xaxis.set_major_formatter(mdates.DateFormatter("%H:%M:%S"))
ax.set_title("KAIROS tick log timeline")
plt.tight_layout()
plt.show()
7. Build a skill library¶
SkillLibrary persists to JSON-lines. Every successful trajectory can be distilled into a reusable skill (name, description, code snippet, tags). Later runs search the library first, so the agent gradually learns the codebase without retraining.
from swarm.skills.library import SkillLibrary
lib = SkillLibrary(library_dir=tempfile.mkdtemp())
await lib.add_skill(
name="debounce",
description="Rate-limit a callback so it only fires after a quiet period.",
code="def debounce(func, wait): ...",
tags=["async", "util"],
)
await lib.add_skill(
name="retry_with_backoff",
description="Call a coroutine until success with exponential backoff.",
code="async def retry(fn, ...): ...",
tags=["async", "reliability"],
)
await lib.add_skill(
name="kv_cache",
description="Small in-process cache with TTL.",
code="class KVCache: ...",
tags=["cache", "performance"],
)
print("library built with 3 skills")
8. Keyword search¶
SkillLibrary.search uses simple word-overlap scoring. It is a placeholder for embedding search — swap in sentence-transformers once you have a real workload and want semantic matches.
hits = await lib.search("async retry on failure")
for h in hits:
print(f"{h.name:20s} tags={h.tags}")
print(f" desc: {h.description[:70]}")
9. Format skills for prompting¶
When a worker needs to know which skills are available, we ship them as a markdown block in the prompt. format_for_prompt renders the library in the exact shape workers expect.
print(lib.format_for_prompt(hits))
10. Plugin directory structure¶
Claude Code plugins live under ~/.claude/plugins/ with a fixed layout. Inspect a real one if available, otherwise show a synthetic example. Colab will not have the plugins directory, so expect the synthetic path.
import os
plugins_root = Path.home() / ".claude" / "plugins"
real_plugins: list[Path] = []
if plugins_root.exists():
try:
real_plugins = [p for p in plugins_root.iterdir() if p.is_dir()]
except Exception as e: # sandboxed env — fall through to synthetic
print(f"Could not list plugins: {e}")
if real_plugins:
print(f"Found {len(real_plugins)} real plugin dirs")
for p in real_plugins[:3]:
print(f"\n{p.name}/")
for entry in sorted(p.rglob("*"))[:8]:
rel = entry.relative_to(p)
print(f" {rel}")
else:
print("No plugins directory available in this environment.")
print("Synthetic example:\n")
example = """my-plugin/
plugin.json # name, version, tools, skills, hooks
skills/
debounce.md # skill docs
commands/
run-tests.md # slash command definition
mcp/
server.py # optional MCP server
hooks/
pre_tool_use.py # before-tool hook
"""
print(example)
11. Distil a skill from a trace¶
The real payoff: after a successful swarm run, pass the execution trace to distill_from_trace. It extracts one reusable skill (if any) and appends to the library. In mock mode the distiller returns a canned example.
sample_trace = """
Agent called fetch_url()
Got 429 Too Many Requests
Retried with 2s sleep - success
Added retry_after header handling
"""
skill = await lib.distill_from_trace(sample_trace, model="claude-haiku-4-5-20251001")
print("distilled:", skill.name if skill else "(no skill extracted)")
12. List all skills after distillation¶
Confirm the library grew (or did not — in mock mode it depends on the fixture). This is the shape you would render in a /skills slash command.
all_skills = lib._load_all()
print(f"library now contains {len(all_skills)} skills:")
for s in all_skills:
print(f" - {s.name:25s} tags={s.tags}")
13. Plot skill tag distribution¶
As a library grows it's useful to know which kinds of skills you have. Tag frequency is a first-pass index.
from collections import Counter
tag_counter = Counter()
for s in all_skills:
tag_counter.update(s.tags)
if tag_counter:
fig, ax = plt.subplots(figsize=(7, 3.5))
ax.bar(tag_counter.keys(), tag_counter.values(), color="#3d7eff")
ax.set_title("Skill tag frequency")
plt.tight_layout()
plt.show()
else:
print("No tags yet.")
14. The daemon + library feedback loop¶
In production, KAIROS ticks periodically. When it dispatches a task, the post-hook feeds the trace back to the skill library. Over weeks the library grows; the next trace starts with a richer prompt. That is the Voyager-style learning curve the field is converging on.
print("daemon log path:", log_path)
print("skill library dir:", lib._dir)
print("wire-up order:")
print(" 1. daemon._tick() picks a pending task")
print(" 2. daemon dispatches it via run_swarm(goal=task)")
print(" 3. post-hook captures the trace")
print(" 4. lib.distill_from_trace(trace) writes a new skill")
print(" 5. next worker sees the skill in lib.format_for_prompt()")
15. What to try next¶
- Start the daemon with
await daemon.start(), wait a few ticks, thenawait daemon.stop(). - Add an event:
daemon.add_event({'type': 'ci_failed', 'repo': 'foo'})and tick. - Add a real API key and run
distill_from_traceon a real swarm trace. - Wire a plugin with a pre_tool_use hook that reads from the skill library.