python3 --version                       # needed for log parsing
echo "${PYLOT_DISPATCH_TOKEN:?PYLOT_DISPATCH_TOKEN not set}"  # auth token required

undefined

All API calls go to `https://hooks.fellowship.dev` with header:

undefined

MISSION=$(curl -sf \
  -H "Authorization: Bearer $PYLOT_DISPATCH_TOKEN" \
  "https://hooks.fellowship.dev/missions/${job_id}")

if [ $? -ne 0 ] || [ -z "$MISSION" ]; then
  echo "ERROR: Failed to fetch mission $job_id"
  exit 1
fi

undefined

curl -sf \
  -H "Authorization: Bearer $PYLOT_DISPATCH_TOKEN" \
  -H "Accept: text/event-stream" \
  "https://hooks.fellowship.dev/missions/${job_id}/logs/stream" \
  > /tmp/distill_logs_${job_id}.txt

echo "Log stream saved."

cat > /tmp/distill_parse_sse.py << 'PYEOF'
import re, sys, json, collections

log_file = sys.argv[1]
signals = {
    "session_complete": True,
    "tool_counts": {},
    "error_tool_results": [],
    "workers_spawned": 0,
    "workers_spawned_sessions": [],
    "skills_loaded": [],
    "verification_calls": [],
    "total_input_tokens": 0,
    "total_output_tokens": 0,
    "original_prompt": None,
    "write_paths": [],
    "bash_calls": [],
    "git_signals": {
        "branch_created": False,
        "pushed": False,
        "push_force": False,
        "pr_created": False,
        "issue_closed": False,
        "committed_to_main": False,
        "push_verified": False
    },
    "timing": {
        "container_start_ts": None,
        "first_claude_call_ts": None,
        "first_token_ts": None,
        "done_signal_ts": None
    }
}
tool_counts = collections.Counter()

try:
    with open(log_file) as f:
        content = f.read()
except Exception as e:
    signals["session_complete"] = False
    print(json.dumps(signals))
    sys.exit(0)

ts = entry.get("timestamp") or entry.get("ts")
msg = entry.get("message", "") or entry.get("msg", "")
msg_lower = msg.lower()

# Timing markers
timing = signals["timing"]
if not timing["container_start_ts"] and any(k in msg_lower for k in [
        "container start", "ecs task started", "bootstrap"]):
    timing["container_start_ts"] = ts
if not timing["first_claude_call_ts"] and any(k in msg_lower for k in [
        "claude api", "anthropic.messages.create", "invoking claude"]):
    timing["first_claude_call_ts"] = ts
if not timing["first_token_ts"] and any(k in msg_lower for k in [
        "first token", "stream started", "content_block_start"]):
    timing["first_token_ts"] = ts
if not timing["done_signal_ts"] and re.search(r'\bdone\b', msg_lower):
    timing["done_signal_ts"] = ts

# Tool call detection from log message text
tool_match = re.search(
    r'\b(Read|Bash|Edit|Write|Grep|Glob|Agent|WebFetch|WebSearch|TodoWrite|TodoRead)\b',
    msg
)
if tool_match:
    tool = tool_match.group(1)
    tool_counts[tool] += 1

    if tool == "Agent":
        signals["workers_spawned"] += 1
    elif tool == "Write":
        path_match = re.search(r'file_path["\s:]+([^\s",}]+)', msg)
        if path_match:
            signals["write_paths"].append(path_match.group(1))
    elif tool == "Read":
        path_match = re.search(r'file_path["\s:]+([^\s",}]+)', msg)
        if path_match and ".claude/skills" in path_match.group(1):
            parts = path_match.group(1).split(".claude/skills/")
            if len(parts) > 1:
                skill_name = parts[1].split("/")[0]
                if skill_name and skill_name not in signals["skills_loaded"]:
                    signals["skills_loaded"].append(skill_name)
    elif tool == "Bash":
        cmd_match = re.search(r'command["\s:]+(.+?)(?:,\s*"|\}|$)', msg)
        if cmd_match:
            cmd = cmd_match.group(1).strip('"')[:300]
            signals["bash_calls"].append(cmd)
            if any(v in cmd for v in ["curl ", "gh pr view", "gh pr merge", "gh pr list", "gh run"]):
                signals["verification_calls"].append(cmd[:200])
            gs = signals["git_signals"]
            if re.search(r'git\s+(checkout\s+-b|switch\s+-c)', cmd):
                gs["branch_created"] = True
            if re.search(r'git\s+push', cmd):
                if "--force" in cmd or " -f " in cmd:
                    gs["push_force"] = True
                gs["pushed"] = True
            if "gh pr create" in cmd:
                gs["pr_created"] = True
            if "gh issue close" in cmd.lower():
                gs["issue_closed"] = True
            if (re.search(r'git\s+log.*--remotes', cmd) or
                    re.search(r'gh\s+api.*commits/', cmd) or
                    re.search(r'git\s+log.*origin/', cmd)):
                gs["push_verified"] = True
            if re.search(r'\bclaude\s+(-p|--print)\b', cmd):
                signals["workers_spawned"] += 1
                sid = re.search(r'--session-id\s+(\S+)', cmd)
                if sid:
                    signals["workers_spawned_sessions"].append(sid.group(1))

# Error patterns — ENOENT, permission denied, timeout, crash
error_patterns = ["enoent", "permission denied", "timeout", "crash", "error:", "exception:"]
if any(p in msg_lower for p in error_patterns):
    signals["error_tool_results"].append(msg[:300])

# Original prompt — first assistant message with task context
if not signals["original_prompt"] and "task:" in msg_lower:
    signals["original_prompt"] = msg[:500]

# Token usage from log lines
tok_in = re.search(r'input_tokens["\s:]+(\d+)', msg)
if tok_in:
    signals["total_input_tokens"] += int(tok_in.group(1))
tok_out = re.search(r'output_tokens["\s:]+(\d+)', msg)
if tok_out:
    signals["total_output_tokens"] += int(tok_out.group(1))

undefined

TIMING_METRICS=$(python3 << PYEOF
import json
from datetime import datetime

signals = json.loads('''$SIGNALS''')
finished_at = "$FINISHED_AT"
timing = signals.get("timing", {})

def parse_ts(ts):
    if not ts:
        return None
    try:
        return datetime.fromisoformat(str(ts).replace("Z", "+00:00"))
    except Exception:
        return None

def ms_between(a, b):
    if a and b:
        return round((b - a).total_seconds() * 1000)
    return None

container_start   = parse_ts(timing.get("container_start_ts"))
first_claude_call = parse_ts(timing.get("first_claude_call_ts"))
first_token       = parse_ts(timing.get("first_token_ts"))
done_signal       = parse_ts(timing.get("done_signal_ts"))
finished          = parse_ts(finished_at)

print(json.dumps({
    "cold_start_ms":    ms_between(container_start, first_claude_call),
    "first_token_ms":   ms_between(first_claude_call, first_token),
    "done_to_ended_ms": ms_between(done_signal, finished)
}))
PYEOF
)

cat > /tmp/distill_classify.py << 'PYEOF'
import json, re, sys

signals = json.loads(sys.argv[1])
report_text = open(sys.argv[2]).read().lower()  # mission report text (from API)

tool_counts = signals.get("tool_counts", {})
bash_calls = signals.get("bash_calls", [])
bash_all = " ".join(bash_calls).lower()
error_results = signals.get("error_tool_results", [])
write_paths = signals.get("write_paths", [])
verification_calls = signals.get("verification_calls", [])
workers_spawned = signals.get("workers_spawned", 0)
skills_loaded = signals.get("skills_loaded", [])
original_prompt = (signals.get("original_prompt") or "").lower()
git_signals = signals.get("git_signals", {})

incidents = []
what_worked = []
missed_opportunity = ""

undefined

HAS_REPORT="True"
[ -z "$REPORT" ] && HAS_REPORT="False"

COMPLIANCE=$(python3 << PYEOF
import json, sys

signals = json.loads('''$SIGNALS''')
task = "$TASK".lower()
tool_counts = signals.get("tool_counts", {})
bash_calls = signals.get("bash_calls", [])
write_paths = signals.get("write_paths", [])
skills_loaded = signals.get("skills_loaded", [])
verification_calls = signals.get("verification_calls", [])
git_signals = signals.get("git_signals", {})

is_feature = any(kw in task for kw in
                 ["implement", "feature", "add", "build", "create", "fix", "update"])

compliance = {
    "R1_headless": "AskUserQuestion" not in tool_counts,
    "R3_no_code": ("Edit" not in tool_counts and
                   not any("src/" in p or "app/" in p for p in write_paths)),
    "R4_read_skills": len(skills_loaded) > 0,
    "R6_verify": len(verification_calls) > 0,
    "R7_report": $HAS_REPORT,  # True if mission API returned a non-empty report field
    "R_branch": git_signals.get("branch_created", False),
    "R_push": git_signals.get("pushed", False),
    "R_pr": git_signals.get("pr_created", False) or not is_feature
}
print(json.dumps(compliance))
PYEOF
)

AUDIT_JSON=$(python3 << PYEOF
import json
from datetime import datetime

signals   = json.loads('''$SIGNALS''')
classified = json.loads('''$CLASSIFIED''')
compliance = json.loads('''$COMPLIANCE''')
timing    = json.loads('''$TIMING_METRICS''')

started_at  = "$STARTED_AT"
finished_at = "$FINISHED_AT"

def parse_ts(ts):
    if not ts:
        return None
    try:
        return datetime.fromisoformat(str(ts).replace("Z", "+00:00"))
    except Exception:
        return None

duration_minutes = 0
s, e = parse_ts(started_at), parse_ts(finished_at)
if s and e:
    duration_minutes = round((e - s).total_seconds() / 60)

input_tok  = signals.get("total_input_tokens", 0)
output_tok = signals.get("total_output_tokens", 0)
cost_usd   = round((input_tok / 1_000_000 * 3.0) + (output_tok / 1_000_000 * 15.0), 2)

audit = {
    "job_id":            "$job_id",
    "team":              "$TEAM",
    "agent":             "$AGENT",
    "task_summary":      "$TASK",
    "status":            "$STATUS",
    "exit_code":         "$EXIT_CODE",
    "started_at":        started_at,
    "finished_at":       finished_at,
    "duration_minutes":  duration_minutes,
    "cost_usd":          cost_usd,
    "token_usage":       {"input": input_tok, "output": output_tok},
    "timing_metrics":    timing,
    "workers_spawned":   signals.get("workers_spawned", 0),
    "skills_loaded":     signals.get("skills_loaded", []),
    "session_complete":  signals.get("session_complete", True),
    "rules_compliance":  compliance,
    "incidents":         classified.get("incidents", []),
    "what_worked":       classified.get("what_worked", []),
    "missed_opportunity": classified.get("missed_opportunity", "")
}
print(json.dumps(audit, indent=2))
PYEOF
)

---

/distill analyze

undefined

All API calls include the header:

undefined

AUTH_HEADER="Authorization: Bearer $PYLOT_DISPATCH_TOKEN"

AUDITS=$(curl -sf \
  -H "$AUTH_HEADER" \
  "https://hooks.fellowship.dev/audits?days=7")

if [ $? -ne 0 ] || [ -z "$AUDITS" ]; then
  echo "ERROR: Failed to fetch audits from hooks.fellowship.dev"
  exit 1
fi

STATS=$(curl -sf \
  -H "$AUTH_HEADER" \
  "https://hooks.fellowship.dev/audits/stats?days=7")

if [ $? -ne 0 ] || [ -z "$STATS" ]; then
  echo "ERROR: Failed to fetch stats from hooks.fellowship.dev"
  exit 1
fi

undefined

cat > /tmp/distill_aggregate.py << 'PYEOF'
import json, sys, collections

audits = json.loads(sys.argv[1])
stats  = json.loads(sys.argv[2])

if isinstance(audits, dict):
    audits = audits.get("records") or audits.get("audits") or []

pattern_counts  = collections.Counter()
pattern_records = collections.defaultdict(list)

for audit in audits:
    skill = audit.get("skill_name") or audit.get("skill") or "unknown"
    team  = audit.get("team", "unknown")
    for inc in audit.get("incidents", []):
        code = inc.get("code", "??")
        key  = (skill, team, code)
        pattern_counts[key] += 1
        pattern_records[key].append({
            "id":          audit.get("job_id") or audit.get("id"),
            "description": inc.get("description", ""),
            "evidence":    inc.get("evidence", "")
        })

recurring = [
    {
        "skill":   k[0],
        "team":    k[1],
        "code":    k[2],
        "count":   pattern_counts[k],
        "records": pattern_records[k]
    }
    for k, cnt in pattern_counts.items()
    if cnt >= 3
]

failure_rates = {
    entry.get("code"): entry.get("rate") or entry.get("failure_rate")
    for entry in (stats.get("by_code") or stats.get("failure_rates") or [])
}

print(json.dumps({
    "total_audits":       len(audits),
    "recurring_patterns": recurring,
    "failure_rates":      failure_rates
}, indent=2))
PYEOF

AGGREGATE=$(python3 /tmp/distill_aggregate.py "$AUDITS" "$STATS")
PATTERN_COUNT=$(python3 -c "import json,sys; print(len(json.load(sys.stdin)['recurring_patterns']))" <<< "$AGGREGATE")
echo "Aggregation complete. ${PATTERN_COUNT} recurring patterns identified."

echo "$AGGREGATE" > /tmp/distill_agg.json
ISSUE_REPO="${ISSUE_REPO:-fellowship-dev/commander}"
python3 << 'PYEOF'
import json, os, subprocess

with open("/tmp/distill_agg.json") as _f:
    agg = json.load(_f)
issue_repo = os.environ.get("ISSUE_REPO", "fellowship-dev/commander")

for pattern in agg.get("recurring_patterns", []):
    skill   = pattern["skill"]
    code    = pattern["code"]
    count   = pattern["count"]
    team    = pattern["team"]
    records = pattern["records"]

    title_prefix = f"[distill] {skill}: recurring {code}"
    title_full   = f"{title_prefix} ({count} occurrences in 7 days)"

    search = subprocess.run(
        ["gh", "issue", "list",
         "--repo", issue_repo,
         "--search", f"{title_prefix} in:title is:open",
         "--json", "number,title", "--limit", "5"],
        capture_output=True, text=True
    )
    existing_issues = json.loads(search.stdout or "[]")
    match = next(
        (i for i in existing_issues if skill in i["title"] and code in i["title"]),
        None
    )

    mission_ids = [r["id"] for r in records if r.get("id")]
    ids_str     = ", ".join(str(m) for m in mission_ids[:5])
    rate        = agg["failure_rates"].get(code)
    rate_str    = f"{rate}%" if isinstance(rate, (int, float)) else str(rate or "N/A")

    if match:
        comment = (
            f"**Updated count**: {count} occurrences in the last 7 days\n\n"
            f"**New exemplar mission IDs**: {ids_str}\n\n"
            f"**Failure rate (7-day)**: {rate_str}"
        )
        result = subprocess.run(
            ["gh", "issue", "comment", str(match["number"]), "--repo", issue_repo, "--body", comment],
            capture_output=True, text=True
        )
        status = "Commented on" if result.returncode == 0 else "WARN: failed to comment on"
        print(f"{status} issue #{match['number']}: {title_prefix}")
    else:
        evidence_lines = "\n".join(
            f"- {r['id']}: {r.get('description', '')[:120]}"
            for r in records[:10]
            if r.get("id")
        )
        body = f"""## Recurring Failure Pattern

**Skill**: \`{skill}\`
**Incident code**: \`{code}\`
**Team**: {team}
**Occurrences**: {count} in the last 7 days
**Failure rate (7-day)**: {rate_str}

undefined

echo "$AGGREGATE" > /tmp/distill_agg.json
python3 << 'PYEOF'
import json, os, urllib.request, urllib.error

token = os.environ.get("PYLOT_DISPATCH_TOKEN", "")
with open("/tmp/distill_agg.json") as _f:
    agg = json.load(_f)

for pattern in agg.get("recurring_patterns", []):
    records = pattern.get("records", [])
    if not records:
        continue

    exemplar = max(
        records,
        key=lambda r: len(r.get("evidence", "") or r.get("description", ""))
    )
    eid = exemplar.get("id")
    if not eid:
        print(f"WARN: No ID for exemplar in {pattern['skill']}/{pattern['code']}, skipping")
        continue

    req = urllib.request.Request(
        f"https://hooks.fellowship.dev/audits/{eid}/exemplar",
        method="PATCH",
        headers={
            "Authorization": f"Bearer {token}",
            "Content-Type":  "application/json"
        },
        data=b"{}"
    )
    try:
        with urllib.request.urlopen(req) as resp:
            print(f"Promoted exemplar {eid} for {pattern['skill']}/{pattern['code']} — HTTP {resp.status}")
    except urllib.error.HTTPError as e:
        print(f"WARN: PATCH /audits/{eid}/exemplar → HTTP {e.code}")
    except Exception as e:
        print(f"WARN: PATCH /audits/{eid}/exemplar → {e}")
PYEOF

echo "$STATS" > /tmp/distill_stats.json
echo "$PRIOR_STATS" > /tmp/distill_prior_stats.json
python3 << 'PYEOF'
import json

with open("/tmp/distill_stats.json") as _f:
    stats = json.load(_f)
with open("/tmp/distill_prior_stats.json") as _f:
    prior_stats = json.load(_f)

def extract_counts(s):
    return {
        e.get("code"): e.get("count", 0)
        for e in (s.get("by_code") or s.get("failure_rates") or [])
    }

current = extract_counts(stats)
prior   = extract_counts(prior_stats)
codes   = sorted(set(current) | set(prior))

print("## Trend Report (current 7 days vs prior 7 days)\n")
print(f"{'Code':<6} {'Current':>9} {'Prior':>9}  Trend")
print("-" * 38)
for code in codes:
    c = current.get(code, 0)
    p = prior.get(code, 0)
    if   c < p: trend = "improving"
    elif c > p: trend = "worsening"
    else:       trend = "stable"
    print(f"{code:<6} {c:>9} {p:>9}  {trend}")
PYEOF

skills:
  - repo: fellowship-dev/dogfooded-skills@latest
    pick: [distill]

cron:
  - schedule: "0 12 * * 1"   # weekly Monday noon
    command: "/distill analyze"

{
  "job_id": "2026-04-12-speckit-fellowship-dev-myrepo-branch",
  "date": "2026-04-12",
  "crew": "tooling",
  "task_summary": "Implement distill skill for post-mission auditing",
  "outcome": "done",
  "session_complete": true,
  "duration_minutes": 47,
  "cost_usd": 3.21,
  "token_usage": { "input": 850000, "output": 45000 },
  "workers_spawned": 2,
  "skills_loaded": ["project-registry", "speckit-runner"],
  "rules_compliance": {
    "R1_headless": true,
    "R3_no_code": true,
    "R4_read_skills": true,
    "R6_verify": true,
    "R7_report": true
  },
  "incidents": [
    {
      "code": "SV",
      "description": "Deploy actions with no HTTP verification",
      "evidence": "17 Bash calls, 0 curl calls"
    }
  ],
  "what_worked": ["Skills loaded on boot", "Worker deployed efficiently"],
  "missed_opportunity": "curl production URL before exit"
}