Compare commits

...

11 Commits

Author SHA1 Message Date
Alexander Whitestone
7399c83024 fix: null guard on assignees in orchestrator dispatch 2026-04-07 12:34:02 -04:00
Alexander Whitestone
cf213bffd1 [SOVEREIGN] Add Orchestrator v1 — backlog reader, priority scorer, agent dispatcher
Resolves #355 #356

Components:
- orchestrator.py: Full sovereign orchestrator with 6 subsystems
  1. Backlog reader (fetches from timmy-config, the-nexus, timmy-home)
  2. Priority scorer (0-100 based on severity, age, assignment state)
  3. Agent roster (groq/ezra/bezalel with health checks)
  4. Dispatcher (matches issues to agents by type/strength)
  5. Consolidated report (terminal + Telegram)
  6. Main loop (--once, --daemon, --dry-run)
- orchestrate.sh: Shell wrapper with env setup

Dry-run tested: 348 issues scanned, 3 agents detected UP.
stdlib only, no pip dependencies.
2026-04-07 12:31:14 -04:00
c1c3aaa681 Merge pull request 'feat: genchi-genbutsu — verify world state, not log vibes (#348)' (#360) from ezra/issue-348 into main 2026-04-07 16:23:35 +00:00
d023512858 Merge pull request 'feat: FLEET-003 - Fleet capacity inventory with resource baselines' (#353) from timmy/fleet-capacity-inventory into main 2026-04-07 16:23:22 +00:00
e5e01e36c9 Merge pull request '[KAIZEN] Automated retrospective after every burn cycle (fixes #349)' (#352) from ezra/issue-349 into main 2026-04-07 16:23:17 +00:00
ezra
e5055d269b feat: genchi-genbutsu — verify world state, not log vibes (#348)
Implement 現地現物 (Genchi Genbutsu) post-completion verification:

- Add bin/genchi-genbutsu.sh performing 5 world-state checks:
  1. Branch exists on remote
  2. PR exists
  3. PR has real file changes (> 0)
  4. PR is mergeable
  5. Issue has a completion comment from the agent

- Wire verification into all agent loops:
  - bin/claude-loop.sh: call genchi-genbutsu before merge/close
  - bin/gemini-loop.sh: delegate existing inline checks to genchi-genbutsu
  - bin/agent-loop.sh: resurrect generic agent loop with genchi-genbutsu wired in

- Update metrics JSONL to include 'verified' field for all loops

- Update burn monitor (tasks.py velocity_tracking):
  - Report verified_completion count alongside raw completions
  - Dashboard shows verified trend history

- Update morning report (tasks.py good_morning_report):
  - Count only verified completions from the last 24h
  - Surface verification failures in the report body

Fixes #348
Refs #345
2026-04-07 16:12:05 +00:00
Alexander Whitestone
277d21aef6 feat: FLEET-007 — Auto-restart agent (self-healing processes)
Daemon that monitors key services and restarts them automatically:
- Local: hermes-gateway, ollama, codeclaw-heartbeat
- Ezra: gitea, nginx, hermes-agent
- Allegro hermes-agent
- Bezalel: hermes-agent, evennia
- Max 3 restart attempts per service per cycle (prevents loops)
- 1-hour cooldown after max retries with Telegram escalation
- Restart log at ~/.local/timmy/fleet-health/restarts.log
- Modes: check now (--status for history, --daemon for continuous)

Fixes timmy-home#560
2026-04-07 12:04:33 -04:00
Alexander Whitestone
228e46a330 feat: FLEET-004/005 — Milestone messages and resource tracker
FLEET-004: 22 milestone messages across 6 phases + 11 Fibonacci uptime milestones.
FLEET-005: Resource tracking system — Capacity/Uptime/Innovation tension model.
  - Tracks capacity spending and regeneration (2/hr baseline)
  - Innovation generates only when utilization < 70% (5/hr scaled)
  - Fibonacci uptime milestone detection (95% through 99.5%)
  - Phase gate checks (P2: 95% uptime, P3: 95% + 100 innovation, P5: 95% + 500)
  - CLI: status, regen commands

Fixes timmy-home#557 (FLEET-004), #558 (FLEET-005)
2026-04-07 12:03:45 -04:00
Ezra
2e64b160b5 [KAIZEN] Harden retro scheduling, chunking, and tests (#349)
- Add Kaizen Retro to cron/jobs.json with explicit local model/provider
- Add Telegram message chunking for reports approaching the 4096-char limit
- Fix classify_issue_type false positives on short substrings (ci in cleanup)
- Add 28 unit tests covering classification, max-attempts detection,
  suggestion generation, report formatting, and Telegram chunking
2026-04-07 15:58:58 +00:00
Alexander Whitestone
67c2927c1a feat: FLEET-003 — Capacity inventory with resource baselines
Full resource audit of all 4 machines (3 VPS + 1 Mac) with:
- vCPU, RAM, disk, swap per machine
- Key processes sorted by resource usage
- Capacity utilization: ~15-20%, Innovation GENERATING
- Uptime baseline: Ezra/Allegro/Bezalel 100%, Gitea 95.8%
- Fibonacci uptime milestones (5 of 6 REACHED)
- Risk assessment (Ezra disk 72%, Bezalel 2GB RAM, Ezra CPU 269%)
- Recommendations across all phases

Fixes timmy-home#556 (FLEET-003)
2026-04-07 11:58:16 -04:00
Ezra
f18955ea90 [KAIZEN] Implement automated burn-cycle retrospective (fixes #349)
- Add bin/kaizen-retro.sh entry point and scripts/kaizen_retro.py
- Analyze closed issues, merged PRs, and stale/max-attempts issues
- Report success rates by agent, repo, and issue type
- Generate one concrete improvement suggestion per cycle
- Post retro to Telegram and comment on the latest morning report issue
- Wire into Huey as kaizen_retro() task at 07:15 daily
- Extend gitea_client.py with since param for list_issues and
  created_at/updated_at fields on PullRequest
2026-04-07 15:57:21 +00:00
17 changed files with 3047 additions and 84 deletions

11
.gitignore vendored
View File

@@ -1,9 +1,8 @@
# Secrets
*.token
*.key
*.secret
# Local state
*.pyc
*.pyo
*.egg-info/
dist/
build/
*.db
*.db-wal
*.db-shm

273
bin/agent-loop.sh Executable file
View File

@@ -0,0 +1,273 @@
#!/usr/bin/env bash
# agent-loop.sh — Universal agent dev loop with Genchi Genbutsu verification
#
# Usage: agent-loop.sh <agent-name> [num-workers]
# agent-loop.sh claude 2
# agent-loop.sh gemini 1
#
# Dispatches via agent-dispatch.sh, then verifies with genchi-genbutsu.sh.
set -uo pipefail
AGENT="${1:?Usage: agent-loop.sh <agent-name> [num-workers]}"
NUM_WORKERS="${2:-1}"
# Resolve agent tool and model from config or fallback
case "$AGENT" in
claude) TOOL="claude"; MODEL="sonnet" ;;
gemini) TOOL="gemini"; MODEL="gemini-2.5-pro-preview-05-06" ;;
grok) TOOL="opencode"; MODEL="grok-3-fast" ;;
*) TOOL="$AGENT"; MODEL="" ;;
esac
# === CONFIG ===
GITEA_URL="${GITEA_URL:-https://forge.alexanderwhitestone.com}"
GITEA_TOKEN="${GITEA_TOKEN:-}"
WORKTREE_BASE="$HOME/worktrees"
LOG_DIR="$HOME/.hermes/logs"
LOCK_DIR="$LOG_DIR/${AGENT}-locks"
SKIP_FILE="$LOG_DIR/${AGENT}-skip-list.json"
ACTIVE_FILE="$LOG_DIR/${AGENT}-active.json"
TIMEOUT=600
COOLDOWN=30
mkdir -p "$LOG_DIR" "$WORKTREE_BASE" "$LOCK_DIR"
[ -f "$SKIP_FILE" ] || echo '{}' > "$SKIP_FILE"
echo '{}' > "$ACTIVE_FILE"
# === SHARED FUNCTIONS ===
log() {
echo "[$(date '+%Y-%m-%d %H:%M:%S')] ${AGENT}: $*" >> "$LOG_DIR/${AGENT}-loop.log"
}
lock_issue() {
local key="$1"
mkdir "$LOCK_DIR/$key.lock" 2>/dev/null && echo $$ > "$LOCK_DIR/$key.lock/pid"
}
unlock_issue() {
rm -rf "$LOCK_DIR/$1.lock" 2>/dev/null
}
mark_skip() {
local issue_num="$1" reason="$2"
python3 -c "
import json, time, fcntl
with open('${SKIP_FILE}', 'r+') as f:
fcntl.flock(f, fcntl.LOCK_EX)
try: skips = json.load(f)
except: skips = {}
failures = skips.get(str($issue_num), {}).get('failures', 0) + 1
skip_hours = 6 if failures >= 3 else 1
skips[str($issue_num)] = {'until': time.time() + (skip_hours * 3600), 'reason': '$reason', 'failures': failures}
f.seek(0); f.truncate()
json.dump(skips, f, indent=2)
" 2>/dev/null
}
get_next_issue() {
python3 -c "
import json, sys, time, urllib.request, os
token = '${GITEA_TOKEN}'
base = '${GITEA_URL}'
repos = ['Timmy_Foundation/the-nexus', 'Timmy_Foundation/timmy-config', 'Timmy_Foundation/hermes-agent']
try:
with open('${SKIP_FILE}') as f: skips = json.load(f)
except: skips = {}
try:
with open('${ACTIVE_FILE}') as f: active = json.load(f); active_issues = {v['issue'] for v in active.values()}
except: active_issues = set()
all_issues = []
for repo in repos:
url = f'{base}/api/v1/repos/{repo}/issues?state=open&type=issues&limit=50&sort=created'
req = urllib.request.Request(url, headers={'Authorization': f'token {token}'})
try:
resp = urllib.request.urlopen(req, timeout=10)
issues = json.loads(resp.read())
for i in issues: i['_repo'] = repo
all_issues.extend(issues)
except: continue
for i in sorted(all_issues, key=lambda x: x['title'].lower()):
assignees = [a['login'] for a in (i.get('assignees') or [])]
if assignees and '${AGENT}' not in assignees: continue
num_str = str(i['number'])
if num_str in active_issues: continue
if skips.get(num_str, {}).get('until', 0) > time.time(): continue
lock = '${LOCK_DIR}/' + i['_repo'].replace('/', '-') + '-' + num_str + '.lock'
if os.path.isdir(lock): continue
owner, name = i['_repo'].split('/')
print(json.dumps({'number': i['number'], 'title': i['title'], 'repo_owner': owner, 'repo_name': name, 'repo': i['_repo']}))
sys.exit(0)
print('null')
" 2>/dev/null
}
# === WORKER FUNCTION ===
run_worker() {
local worker_id="$1"
log "WORKER-${worker_id}: Started"
while true; do
issue_json=$(get_next_issue)
if [ "$issue_json" = "null" ] || [ -z "$issue_json" ]; then
sleep 30
continue
fi
issue_num=$(echo "$issue_json" | python3 -c "import sys,json; print(json.load(sys.stdin)['number'])")
issue_title=$(echo "$issue_json" | python3 -c "import sys,json; print(json.load(sys.stdin)['title'])")
repo_owner=$(echo "$issue_json" | python3 -c "import sys,json; print(json.load(sys.stdin)['repo_owner'])")
repo_name=$(echo "$issue_json" | python3 -c "import sys,json; print(json.load(sys.stdin)['repo_name'])")
issue_key="${repo_owner}-${repo_name}-${issue_num}"
branch="${AGENT}/issue-${issue_num}"
worktree="${WORKTREE_BASE}/${AGENT}-w${worker_id}-${issue_num}"
if ! lock_issue "$issue_key"; then
sleep 5
continue
fi
log "WORKER-${worker_id}: === ISSUE #${issue_num}: ${issue_title} (${repo_owner}/${repo_name}) ==="
# Clone / checkout
rm -rf "$worktree" 2>/dev/null
CLONE_URL="http://${AGENT}:${GITEA_TOKEN}@143.198.27.163:3000/${repo_owner}/${repo_name}.git"
if git ls-remote --heads "$CLONE_URL" "$branch" 2>/dev/null | grep -q "$branch"; then
git clone --depth=50 -b "$branch" "$CLONE_URL" "$worktree" >/dev/null 2>&1
else
git clone --depth=1 -b main "$CLONE_URL" "$worktree" >/dev/null 2>&1
cd "$worktree" && git checkout -b "$branch" >/dev/null 2>&1
fi
cd "$worktree"
# Generate prompt
prompt=$(bash "$(dirname "$0")/agent-dispatch.sh" "$AGENT" "$issue_num" "${repo_owner}/${repo_name}")
CYCLE_START=$(date +%s)
set +e
if [ "$TOOL" = "claude" ]; then
env -u CLAUDECODE gtimeout "$TIMEOUT" claude \
--print --model "$MODEL" --dangerously-skip-permissions \
-p "$prompt" </dev/null >> "$LOG_DIR/${AGENT}-${issue_num}.log" 2>&1
elif [ "$TOOL" = "gemini" ]; then
gtimeout "$TIMEOUT" gemini -p "$prompt" --yolo \
</dev/null >> "$LOG_DIR/${AGENT}-${issue_num}.log" 2>&1
else
gtimeout "$TIMEOUT" "$TOOL" "$prompt" \
</dev/null >> "$LOG_DIR/${AGENT}-${issue_num}.log" 2>&1
fi
exit_code=$?
set -e
CYCLE_END=$(date +%s)
CYCLE_DURATION=$((CYCLE_END - CYCLE_START))
# Salvage
cd "$worktree" 2>/dev/null || true
DIRTY=$(git status --porcelain 2>/dev/null | wc -l | tr -d ' ')
if [ "${DIRTY:-0}" -gt 0 ]; then
git add -A 2>/dev/null
git commit -m "WIP: ${AGENT} progress on #${issue_num}
Automated salvage commit — agent session ended (exit $exit_code)." 2>/dev/null || true
fi
UNPUSHED=$(git log --oneline "origin/main..HEAD" 2>/dev/null | wc -l | tr -d ' ')
if [ "${UNPUSHED:-0}" -gt 0 ]; then
git push -u origin "$branch" 2>/dev/null && \
log "WORKER-${worker_id}: Pushed $UNPUSHED commit(s) on $branch" || \
log "WORKER-${worker_id}: Push failed for $branch"
fi
# Create PR if needed
pr_num=$(curl -sf "${GITEA_URL}/api/v1/repos/${repo_owner}/${repo_name}/pulls?state=open&head=${repo_owner}:${branch}&limit=1" \
-H "Authorization: token ${GITEA_TOKEN}" | python3 -c "
import sys,json
prs = json.load(sys.stdin)
print(prs[0]['number'] if prs else '')
" 2>/dev/null)
if [ -z "$pr_num" ] && [ "${UNPUSHED:-0}" -gt 0 ]; then
pr_num=$(curl -sf -X POST "${GITEA_URL}/api/v1/repos/${repo_owner}/${repo_name}/pulls" \
-H "Authorization: token ${GITEA_TOKEN}" \
-H "Content-Type: application/json" \
-d "$(python3 -c "
import json
print(json.dumps({
'title': '${AGENT}: Issue #${issue_num}',
'head': '${branch}',
'base': 'main',
'body': 'Automated PR for issue #${issue_num}.\nExit code: ${exit_code}'
}))
")" | python3 -c "import sys,json; print(json.load(sys.stdin).get('number',''))" 2>/dev/null)
[ -n "$pr_num" ] && log "WORKER-${worker_id}: Created PR #${pr_num} for issue #${issue_num}"
fi
# ── Genchi Genbutsu: verify world state before declaring success ──
VERIFIED="false"
if [ "$exit_code" -eq 0 ]; then
log "WORKER-${worker_id}: SUCCESS #${issue_num} — running genchi-genbutsu"
SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
if verify_result=$("$SCRIPT_DIR/genchi-genbutsu.sh" "$repo_owner" "$repo_name" "$issue_num" "$branch" "$AGENT" 2>/dev/null); then
VERIFIED="true"
log "WORKER-${worker_id}: VERIFIED #${issue_num}"
if [ -n "$pr_num" ]; then
curl -sf -X POST "${GITEA_URL}/api/v1/repos/${repo_owner}/${repo_name}/pulls/${pr_num}/merge" \
-H "Authorization: token ${GITEA_TOKEN}" \
-H "Content-Type: application/json" \
-d '{"Do": "squash"}' >/dev/null 2>&1 || true
curl -sf -X PATCH "${GITEA_URL}/api/v1/repos/${repo_owner}/${repo_name}/issues/${issue_num}" \
-H "Authorization: token ${GITEA_TOKEN}" \
-H "Content-Type: application/json" \
-d '{"state": "closed"}' >/dev/null 2>&1 || true
log "WORKER-${worker_id}: PR #${pr_num} merged, issue #${issue_num} closed"
fi
consecutive_failures=0
else
verify_details=$(echo "$verify_result" | python3 -c "import sys,json; print(json.load(sys.stdin).get('details','unknown'))" 2>/dev/null || echo "unverified")
log "WORKER-${worker_id}: UNVERIFIED #${issue_num}$verify_details"
mark_skip "$issue_num" "unverified" 1
consecutive_failures=$((consecutive_failures + 1))
fi
elif [ "$exit_code" -eq 124 ]; then
log "WORKER-${worker_id}: TIMEOUT #${issue_num} (work saved in PR)"
consecutive_failures=$((consecutive_failures + 1))
else
log "WORKER-${worker_id}: FAILED #${issue_num} exit ${exit_code} (work saved in PR)"
consecutive_failures=$((consecutive_failures + 1))
fi
# ── METRICS ──
python3 -c "
import json, datetime
print(json.dumps({
'ts': datetime.datetime.utcnow().isoformat() + 'Z',
'agent': '${AGENT}',
'worker': $worker_id,
'issue': $issue_num,
'repo': '${repo_owner}/${repo_name}',
'outcome': 'success' if $exit_code == 0 else 'timeout' if $exit_code == 124 else 'failed',
'exit_code': $exit_code,
'duration_s': $CYCLE_DURATION,
'pr': '${pr_num:-}',
'verified': ${VERIFIED:-false}
}))
" >> "$LOG_DIR/${AGENT}-metrics.jsonl" 2>/dev/null
rm -rf "$worktree" 2>/dev/null
unlock_issue "$issue_key"
sleep "$COOLDOWN"
done
}
# === MAIN ===
log "=== Agent Loop Started — ${AGENT} with ${NUM_WORKERS} worker(s) ==="
rm -rf "$LOCK_DIR"/*.lock 2>/dev/null
for i in $(seq 1 "$NUM_WORKERS"); do
run_worker "$i" &
log "Launched worker $i (PID $!)"
sleep 3
done
wait

View File

@@ -468,24 +468,32 @@ print(json.dumps({
[ -n "$pr_num" ] && log "WORKER-${worker_id}: Created PR #${pr_num} for issue #${issue_num}"
fi
# ── Merge + close on success ──
# ── Genchi Genbutsu: verify world state before declaring success ──
VERIFIED="false"
if [ "$exit_code" -eq 0 ]; then
log "WORKER-${worker_id}: SUCCESS #${issue_num}"
if [ -n "$pr_num" ]; then
curl -sf -X POST "${GITEA_URL}/api/v1/repos/${repo_owner}/${repo_name}/pulls/${pr_num}/merge" \
-H "Authorization: token ${GITEA_TOKEN}" \
-H "Content-Type: application/json" \
-d '{"Do": "squash"}' >/dev/null 2>&1 || true
curl -sf -X PATCH "${GITEA_URL}/api/v1/repos/${repo_owner}/${repo_name}/issues/${issue_num}" \
-H "Authorization: token ${GITEA_TOKEN}" \
-H "Content-Type: application/json" \
-d '{"state": "closed"}' >/dev/null 2>&1 || true
log "WORKER-${worker_id}: PR #${pr_num} merged, issue #${issue_num} closed"
log "WORKER-${worker_id}: SUCCESS #${issue_num} — running genchi-genbutsu"
SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
if verify_result=$("$SCRIPT_DIR/genchi-genbutsu.sh" "$repo_owner" "$repo_name" "$issue_num" "$branch" "claude" 2>/dev/null); then
VERIFIED="true"
log "WORKER-${worker_id}: VERIFIED #${issue_num}"
if [ -n "$pr_num" ]; then
curl -sf -X POST "${GITEA_URL}/api/v1/repos/${repo_owner}/${repo_name}/pulls/${pr_num}/merge" \
-H "Authorization: token ${GITEA_TOKEN}" \
-H "Content-Type: application/json" \
-d '{"Do": "squash"}' >/dev/null 2>&1 || true
curl -sf -X PATCH "${GITEA_URL}/api/v1/repos/${repo_owner}/${repo_name}/issues/${issue_num}" \
-H "Authorization: token ${GITEA_TOKEN}" \
-H "Content-Type: application/json" \
-d '{"state": "closed"}' >/dev/null 2>&1 || true
log "WORKER-${worker_id}: PR #${pr_num} merged, issue #${issue_num} closed"
fi
consecutive_failures=0
else
verify_details=$(echo "$verify_result" | python3 -c "import sys,json; print(json.load(sys.stdin).get('details','unknown'))" 2>/dev/null || echo "unverified")
log "WORKER-${worker_id}: UNVERIFIED #${issue_num}$verify_details"
consecutive_failures=$((consecutive_failures + 1))
fi
consecutive_failures=0
elif [ "$exit_code" -eq 124 ]; then
log "WORKER-${worker_id}: TIMEOUT #${issue_num} (work saved in PR)"
consecutive_failures=$((consecutive_failures + 1))
@@ -522,6 +530,7 @@ print(json.dumps({
import json, datetime
print(json.dumps({
'ts': datetime.datetime.utcnow().isoformat() + 'Z',
'agent': 'claude',
'worker': $worker_id,
'issue': $issue_num,
'repo': '${repo_owner}/${repo_name}',
@@ -534,7 +543,8 @@ print(json.dumps({
'lines_removed': ${LINES_REMOVED:-0},
'salvaged': ${DIRTY:-0},
'pr': '${pr_num:-}',
'merged': $( [ '$OUTCOME' = 'success' ] && [ -n '${pr_num:-}' ] && echo 'true' || echo 'false' )
'merged': $( [ '$OUTCOME' = 'success' ] && [ -n '${pr_num:-}' ] && echo 'true' || echo 'false' ),
'verified': ${VERIFIED:-false}
}))
" >> "$METRICS_FILE" 2>/dev/null

View File

@@ -521,61 +521,63 @@ print(json.dumps({
[ -n "$pr_num" ] && log "WORKER-${worker_id}: Created PR #${pr_num} for issue #${issue_num}"
fi
# ── Verify finish semantics / classify failures ──
# ── Genchi Genbutsu: verify world state before declaring success ──
VERIFIED="false"
if [ "$exit_code" -eq 0 ]; then
log "WORKER-${worker_id}: SUCCESS #${issue_num} exited 0 — verifying push + PR + proof"
if ! remote_branch_exists "$branch"; then
log "WORKER-${worker_id}: BLOCKED #${issue_num} remote branch missing"
post_issue_comment "$repo_owner" "$repo_name" "$issue_num" "Loop gate blocked completion: remote branch ${branch} was not found on origin after Gemini exited. Issue remains open for retry."
mark_skip "$issue_num" "missing_remote_branch" 1
consecutive_failures=$((consecutive_failures + 1))
elif [ -z "$pr_num" ]; then
log "WORKER-${worker_id}: BLOCKED #${issue_num} no PR found"
post_issue_comment "$repo_owner" "$repo_name" "$issue_num" "Loop gate blocked completion: branch ${branch} exists remotely, but no PR was found. Issue remains open for retry."
mark_skip "$issue_num" "missing_pr" 1
consecutive_failures=$((consecutive_failures + 1))
log "WORKER-${worker_id}: SUCCESS #${issue_num} exited 0 — running genchi-genbutsu"
SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
if verify_result=$("$SCRIPT_DIR/genchi-genbutsu.sh" "$repo_owner" "$repo_name" "$issue_num" "$branch" "gemini" 2>/dev/null); then
VERIFIED="true"
log "WORKER-${worker_id}: VERIFIED #${issue_num}"
pr_state=$(get_pr_state "$repo_owner" "$repo_name" "$pr_num")
if [ "$pr_state" = "open" ]; then
curl -sf -X POST "${GITEA_URL}/api/v1/repos/${repo_owner}/${repo_name}/pulls/${pr_num}/merge" \
-H "Authorization: token ${GITEA_TOKEN}" \
-H "Content-Type: application/json" \
-d '{"Do": "squash"}' >/dev/null 2>&1 || true
pr_state=$(get_pr_state "$repo_owner" "$repo_name" "$pr_num")
fi
if [ "$pr_state" = "merged" ]; then
curl -sf -X PATCH "${GITEA_URL}/api/v1/repos/${repo_owner}/${repo_name}/issues/${issue_num}" \
-H "Authorization: token ${GITEA_TOKEN}" \
-H "Content-Type: application/json" \
-d '{"state": "closed"}' >/dev/null 2>&1 || true
issue_state=$(get_issue_state "$repo_owner" "$repo_name" "$issue_num")
if [ "$issue_state" = "closed" ]; then
log "WORKER-${worker_id}: VERIFIED #${issue_num} branch pushed, PR merged, comment present, issue closed"
consecutive_failures=0
else
log "WORKER-${worker_id}: BLOCKED #${issue_num} issue did not close after merge"
mark_skip "$issue_num" "issue_close_unverified" 1
consecutive_failures=$((consecutive_failures + 1))
fi
else
log "WORKER-${worker_id}: BLOCKED #${issue_num} merge not verified (state=${pr_state})"
mark_skip "$issue_num" "merge_unverified" 1
consecutive_failures=$((consecutive_failures + 1))
fi
else
pr_files=$(get_pr_file_count "$repo_owner" "$repo_name" "$pr_num")
if [ "${pr_files:-0}" -eq 0 ]; then
log "WORKER-${worker_id}: BLOCKED #${issue_num} PR #${pr_num} has 0 changed files"
curl -sf -X PATCH "${GITEA_URL}/api/v1/repos/${repo_owner}/${repo_name}/pulls/${pr_num}" -H "Authorization: token ${GITEA_TOKEN}" -H "Content-Type: application/json" -d '{"state": "closed"}' >/dev/null 2>&1 || true
verify_details=$(echo "$verify_result" | python3 -c "import sys,json; print(json.load(sys.stdin).get('details','unknown'))" 2>/dev/null || echo "unverified")
verify_checks=$(echo "$verify_result" | python3 -c "import sys,json; print(json.load(sys.stdin).get('checks',''))" 2>/dev/null || echo "")
log "WORKER-${worker_id}: UNVERIFIED #${issue_num} $verify_details"
if echo "$verify_checks" | grep -q '"branch": false'; then
post_issue_comment "$repo_owner" "$repo_name" "$issue_num" "Loop gate blocked completion: remote branch ${branch} was not found on origin after Gemini exited. Issue remains open for retry."
mark_skip "$issue_num" "missing_remote_branch" 1
elif echo "$verify_checks" | grep -q '"pr": false'; then
post_issue_comment "$repo_owner" "$repo_name" "$issue_num" "Loop gate blocked completion: branch ${branch} exists remotely, but no PR was found. Issue remains open for retry."
mark_skip "$issue_num" "missing_pr" 1
elif echo "$verify_checks" | grep -q '"files": false'; then
curl -sf -X PATCH "${GITEA_URL}/api/v1/repos/${repo_owner}/${repo_name}/pulls/${pr_num}" \
-H "Authorization: token ${GITEA_TOKEN}" \
-H "Content-Type: application/json" \
-d '{"state": "closed"}' >/dev/null 2>&1 || true
post_issue_comment "$repo_owner" "$repo_name" "$issue_num" "PR #${pr_num} was closed automatically: it had 0 changed files (empty commit). Issue remains open for retry."
mark_skip "$issue_num" "empty_commit" 2
consecutive_failures=$((consecutive_failures + 1))
else
proof_status=$(proof_comment_status "$repo_owner" "$repo_name" "$issue_num" "$branch")
proof_state="${proof_status%%|*}"
proof_url="${proof_status#*|}"
if [ "$proof_state" != "ok" ]; then
log "WORKER-${worker_id}: BLOCKED #${issue_num} proof missing or incomplete (${proof_state})"
post_issue_comment "$repo_owner" "$repo_name" "$issue_num" "Loop gate blocked completion: PR #${pr_num} exists and has ${pr_files} changed file(s), but the required Proof block from Gemini is missing or incomplete. Issue remains open for retry."
mark_skip "$issue_num" "missing_proof" 1
consecutive_failures=$((consecutive_failures + 1))
else
log "WORKER-${worker_id}: PROOF verified ${proof_url}"
pr_state=$(get_pr_state "$repo_owner" "$repo_name" "$pr_num")
if [ "$pr_state" = "open" ]; then
curl -sf -X POST "${GITEA_URL}/api/v1/repos/${repo_owner}/${repo_name}/pulls/${pr_num}/merge" -H "Authorization: token ${GITEA_TOKEN}" -H "Content-Type: application/json" -d '{"Do": "squash"}' >/dev/null 2>&1 || true
pr_state=$(get_pr_state "$repo_owner" "$repo_name" "$pr_num")
fi
if [ "$pr_state" = "merged" ]; then
curl -sf -X PATCH "${GITEA_URL}/api/v1/repos/${repo_owner}/${repo_name}/issues/${issue_num}" -H "Authorization: token ${GITEA_TOKEN}" -H "Content-Type: application/json" -d '{"state": "closed"}' >/dev/null 2>&1 || true
issue_state=$(get_issue_state "$repo_owner" "$repo_name" "$issue_num")
if [ "$issue_state" = "closed" ]; then
log "WORKER-${worker_id}: VERIFIED #${issue_num} branch pushed, PR merged, proof present, issue closed"
consecutive_failures=0
else
log "WORKER-${worker_id}: BLOCKED #${issue_num} issue did not close after merge"
mark_skip "$issue_num" "issue_close_unverified" 1
consecutive_failures=$((consecutive_failures + 1))
fi
else
log "WORKER-${worker_id}: BLOCKED #${issue_num} merge not verified (state=${pr_state})"
mark_skip "$issue_num" "merge_unverified" 1
consecutive_failures=$((consecutive_failures + 1))
fi
fi
post_issue_comment "$repo_owner" "$repo_name" "$issue_num" "Loop gate blocked completion: PR #${pr_num} exists, but required verification failed ($verify_details). Issue remains open for retry."
mark_skip "$issue_num" "unverified" 1
fi
consecutive_failures=$((consecutive_failures + 1))
fi
elif [ "$exit_code" -eq 124 ]; then
log "WORKER-${worker_id}: TIMEOUT #${issue_num} (work saved in PR)"
@@ -621,7 +623,8 @@ print(json.dumps({
'lines_removed': ${LINES_REMOVED:-0},
'salvaged': ${DIRTY:-0},
'pr': '${pr_num:-}',
'merged': $( [ '$OUTCOME' = 'success' ] && [ -n '${pr_num:-}' ] && echo 'true' || echo 'false' )
'merged': $( [ '$OUTCOME' = 'success' ] && [ -n '${pr_num:-}' ] && echo 'true' || echo 'false' ),
'verified': ${VERIFIED:-false}
}))
" >> "$LOG_DIR/gemini-metrics.jsonl" 2>/dev/null

179
bin/genchi-genbutsu.sh Executable file
View File

@@ -0,0 +1,179 @@
#!/usr/bin/env bash
# genchi-genbutsu.sh — 現地現物 — Go and see. Verify world state, not log vibes.
#
# Post-completion verification that goes and LOOKS at the actual artifacts.
# Performs 5 world-state checks:
# 1. Branch exists on remote
# 2. PR exists
# 3. PR has real file changes (> 0)
# 4. PR is mergeable
# 5. Issue has a completion comment from the agent
#
# Usage: genchi-genbutsu.sh <repo_owner> <repo_name> <issue_num> <branch> <agent_name>
# Returns: JSON to stdout, logs JSONL, exit 0 = VERIFIED, exit 1 = UNVERIFIED
set -euo pipefail
GITEA_URL="${GITEA_URL:-https://forge.alexanderwhitestone.com}"
GITEA_TOKEN="${GITEA_TOKEN:-}"
LOG_DIR="${LOG_DIR:-$HOME/.hermes/logs}"
VERIFY_LOG="$LOG_DIR/genchi-genbutsu.jsonl"
if [ $# -lt 5 ]; then
echo "Usage: $0 <repo_owner> <repo_name> <issue_num> <branch> <agent_name>" >&2
exit 2
fi
repo_owner="$1"
repo_name="$2"
issue_num="$3"
branch="$4"
agent_name="$5"
mkdir -p "$LOG_DIR"
# ── Helpers ──────────────────────────────────────────────────────────
check_branch_exists() {
# Use Gitea API instead of git ls-remote so we don't need clone credentials
curl -sf "${GITEA_URL}/api/v1/repos/${repo_owner}/${repo_name}/branches/${branch}" \
-H "Authorization: token ${GITEA_TOKEN}" >/dev/null 2>&1
}
get_pr_num() {
curl -sf "${GITEA_URL}/api/v1/repos/${repo_owner}/${repo_name}/pulls?state=all&head=${repo_owner}:${branch}&limit=1" \
-H "Authorization: token ${GITEA_TOKEN}" 2>/dev/null | python3 -c "
import sys, json
prs = json.load(sys.stdin)
print(prs[0]['number'] if prs else '')
"
}
check_pr_files() {
local pr_num="$1"
curl -sf "${GITEA_URL}/api/v1/repos/${repo_owner}/${repo_name}/pulls/${pr_num}/files" \
-H "Authorization: token ${GITEA_TOKEN}" 2>/dev/null | python3 -c "
import sys, json
try:
files = json.load(sys.stdin)
print(len(files) if isinstance(files, list) else 0)
except:
print(0)
"
}
check_pr_mergeable() {
local pr_num="$1"
curl -sf "${GITEA_URL}/api/v1/repos/${repo_owner}/${repo_name}/pulls/${pr_num}" \
-H "Authorization: token ${GITEA_TOKEN}" 2>/dev/null | python3 -c "
import sys, json
pr = json.load(sys.stdin)
print('true' if pr.get('mergeable') else 'false')
"
}
check_completion_comment() {
curl -sf "${GITEA_URL}/api/v1/repos/${repo_owner}/${repo_name}/issues/${issue_num}/comments" \
-H "Authorization: token ${GITEA_TOKEN}" 2>/dev/null | AGENT="$agent_name" python3 -c "
import os, sys, json
agent = os.environ.get('AGENT', '').lower()
try:
comments = json.load(sys.stdin)
except:
sys.exit(1)
for c in reversed(comments):
user = ((c.get('user') or {}).get('login') or '').lower()
if user == agent:
sys.exit(0)
sys.exit(1)
"
}
# ── Run checks ───────────────────────────────────────────────────────
ts=$(date -u '+%Y-%m-%dT%H:%M:%SZ')
status="VERIFIED"
details=()
checks_json='{}'
# Check 1: branch
if check_branch_exists; then
checks_json=$(echo "$checks_json" | python3 -c "import sys,json;d=json.load(sys.stdin);d['branch']=True;print(json.dumps(d))")
else
checks_json=$(echo "$checks_json" | python3 -c "import sys,json;d=json.load(sys.stdin);d['branch']=False;print(json.dumps(d))")
status="UNVERIFIED"
details+=("remote branch ${branch} not found")
fi
# Check 2: PR exists
pr_num=$(get_pr_num)
if [ -n "$pr_num" ]; then
checks_json=$(echo "$checks_json" | python3 -c "import sys,json;d=json.load(sys.stdin);d['pr']=True;print(json.dumps(d))")
else
checks_json=$(echo "$checks_json" | python3 -c "import sys,json;d=json.load(sys.stdin);d['pr']=False;print(json.dumps(d))")
status="UNVERIFIED"
details+=("no PR found for branch ${branch}")
fi
# Check 3: PR has real file changes
if [ -n "$pr_num" ]; then
file_count=$(check_pr_files "$pr_num")
if [ "${file_count:-0}" -gt 0 ]; then
checks_json=$(echo "$checks_json" | python3 -c "import sys,json;d=json.load(sys.stdin);d['files']=True;print(json.dumps(d))")
else
checks_json=$(echo "$checks_json" | python3 -c "import sys,json;d=json.load(sys.stdin);d['files']=False;print(json.dumps(d))")
status="UNVERIFIED"
details+=("PR #${pr_num} has 0 changed files")
fi
# Check 4: PR is mergeable
if [ "$(check_pr_mergeable "$pr_num")" = "true" ]; then
checks_json=$(echo "$checks_json" | python3 -c "import sys,json;d=json.load(sys.stdin);d['mergeable']=True;print(json.dumps(d))")
else
checks_json=$(echo "$checks_json" | python3 -c "import sys,json;d=json.load(sys.stdin);d['mergeable']=False;print(json.dumps(d))")
status="UNVERIFIED"
details+=("PR #${pr_num} is not mergeable")
fi
else
checks_json=$(echo "$checks_json" | python3 -c "import sys,json;d=json.load(sys.stdin);d['files']=None;d['mergeable']=None;print(json.dumps(d))")
fi
# Check 5: completion comment from agent
if check_completion_comment; then
checks_json=$(echo "$checks_json" | python3 -c "import sys,json;d=json.load(sys.stdin);d['comment']=True;print(json.dumps(d))")
else
checks_json=$(echo "$checks_json" | python3 -c "import sys,json;d=json.load(sys.stdin);d['comment']=False;print(json.dumps(d))")
status="UNVERIFIED"
details+=("no completion comment from ${agent_name} on issue #${issue_num}")
fi
# Build detail string
detail_str=$(IFS="; "; echo "${details[*]:-all checks passed}")
# ── Output ───────────────────────────────────────────────────────────
result=$(python3 -c "
import json
print(json.dumps({
'status': '$status',
'repo': '${repo_owner}/${repo_name}',
'issue': $issue_num,
'branch': '$branch',
'agent': '$agent_name',
'pr': '$pr_num',
'checks': $checks_json,
'details': '$detail_str',
'ts': '$ts'
}, indent=2))
")
printf '%s\n' "$result"
# Append to JSONL log
printf '%s\n' "$result" >> "$VERIFY_LOG"
if [ "$status" = "VERIFIED" ]; then
exit 0
else
exit 1
fi

45
bin/kaizen-retro.sh Executable file
View File

@@ -0,0 +1,45 @@
#!/usr/bin/env bash
# kaizen-retro.sh — Automated retrospective after every burn cycle.
#
# Runs daily after the morning report.
# Analyzes success rates by agent, repo, and issue type.
# Identifies max-attempts issues, generates ONE concrete improvement,
# and posts the retro to Telegram + the master morning-report issue.
#
# Usage:
# ./bin/kaizen-retro.sh [--dry-run]
set -euo pipefail
SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
REPO_ROOT="${SCRIPT_DIR%/bin}"
PYTHON="${PYTHON3:-python3}"
# Source local env if available so TELEGRAM_BOT_TOKEN is picked up
HOME_DIR="${HOME:-$(eval echo ~$(whoami))}"
for env_file in "$HOME_DIR/.hermes/.env" "$HOME_DIR/.timmy/.env" "$REPO_ROOT/.env"; do
if [ -f "$env_file" ]; then
# shellcheck source=/dev/null
set -a
# shellcheck source=/dev/null
source "$env_file"
set +a
fi
done
# If the configured Gitea URL is unreachable but localhost works, prefer localhost
if ! curl -sf "${GITEA_URL:-http://localhost:3000}/api/v1/version" >/dev/null 2>&1; then
if curl -sf http://localhost:3000/api/v1/version >/dev/null 2>&1; then
export GITEA_URL="http://localhost:3000"
fi
fi
# Ensure the Python script exists
RETRO_PY="$REPO_ROOT/scripts/kaizen_retro.py"
if [ ! -f "$RETRO_PY" ]; then
echo "ERROR: kaizen_retro.py not found at $RETRO_PY" >&2
exit 1
fi
# Run
exec "$PYTHON" "$RETRO_PY" "$@"

View File

@@ -137,7 +137,38 @@
"paused_reason": null,
"skills": [],
"skill": null
},
{
"id": "kaizen-retro-349",
"name": "Kaizen Retro",
"prompt": "Run the automated burn-cycle retrospective. Execute: cd /root/wizards/ezra/workspace/timmy-config && ./bin/kaizen-retro.sh",
"model": "hermes3:latest",
"provider": "ollama",
"base_url": "http://localhost:11434/v1",
"schedule": {
"kind": "interval",
"minutes": 1440,
"display": "every 1440m"
},
"schedule_display": "daily at 07:30",
"repeat": {
"times": null,
"completed": 0
},
"enabled": true,
"created_at": "2026-04-07T15:30:00.000000Z",
"next_run_at": "2026-04-08T07:30:00.000000Z",
"last_run_at": null,
"last_status": null,
"last_error": null,
"deliver": "local",
"origin": null,
"state": "scheduled",
"paused_at": null,
"paused_reason": null,
"skills": [],
"skill": null
}
],
"updated_at": "2026-04-07T15:00:00+00:00"
}
}

272
fleet/auto_restart.py Executable file
View File

@@ -0,0 +1,272 @@
#!/usr/bin/env python3
"""
Auto-Restart Agent — Self-healing process monitor for fleet machines.
Detects dead services and restarts them automatically.
Escalates after 3 attempts (prevents restart loops).
Logs all actions to ~/.local/timmy/fleet-health/restarts.log
Alerts via Telegram if service cannot be recovered.
Prerequisite: FLEET-006 (health check) must be running to detect failures.
Usage:
python3 auto_restart.py # Run checks now
python3 auto_restart.py --daemon # Run continuously (every 60s)
python3 auto_restart.py --status # Show restart history
"""
import os
import sys
import json
import time
import subprocess
from datetime import datetime, timezone
from pathlib import Path
# === CONFIG ===
LOG_DIR = Path(os.path.expanduser("~/.local/timmy/fleet-health"))
RESTART_LOG = LOG_DIR / "restarts.log"
COOLDOWN_FILE = LOG_DIR / "restart_cooldowns.json"
MAX_RETRIES = 3
COOLDOWN_PERIOD = 3600 # 1 hour between escalation alerts
# Services definition: name, check command, restart command
# Local services:
LOCAL_SERVICES = {
"hermes-gateway": {
"check": "pgrep -f 'hermes gateway' > /dev/null 2>/dev/null",
"restart": "cd ~/code-claw && ./restart-gateway.sh 2>/dev/null || launchctl kickstart -k ai.hermes.gateway 2>/dev/null",
"critical": True,
},
"ollama": {
"check": "pgrep -f 'ollama serve' > /dev/null 2>/dev/null",
"restart": "launchctl kickstart -k com.ollama.ollama 2>/dev/null || /opt/homebrew/bin/brew services restart ollama 2>/dev/null",
"critical": False,
},
"codeclaw-heartbeat": {
"check": "launchctl list | grep 'ai.timmy.codeclaw-qwen-heartbeat' > /dev/null 2>/dev/null",
"restart": "launchctl kickstart -k ai.timmy.codeclaw-qwen-heartbeat 2>/dev/null",
"critical": False,
},
}
# VPS services to restart via SSH
VPS_SERVICES = {
"ezra": {
"ip": "143.198.27.163",
"user": "root",
"services": {
"gitea": {
"check": "systemctl is-active gitea 2>/dev/null | grep -q active",
"restart": "systemctl restart gitea 2>/dev/null",
"critical": True,
},
"nginx": {
"check": "systemctl is-active nginx 2>/dev/null | grep -q active",
"restart": "systemctl restart nginx 2>/dev/null",
"critical": False,
},
"hermes-agent": {
"check": "pgrep -f 'hermes gateway' > /dev/null 2>/dev/null",
"restart": "cd /root/wizards/ezra/hermes-agent && source .venv/bin/activate && nohup hermes gateway run --replace > /dev/null 2>&1 &",
"critical": True,
},
},
},
"allegro": {
"ip": "167.99.126.228",
"user": "root",
"services": {
"hermes-agent": {
"check": "pgrep -f 'hermes gateway' > /dev/null 2>/dev/null",
"restart": "cd /root/wizards/allegro/hermes-agent && source .venv/bin/activate && nohup hermes gateway run --replace > /dev/null 2>&1 &",
"critical": True,
},
},
},
"bezalel": {
"ip": "159.203.146.185",
"user": "root",
"services": {
"hermes-agent": {
"check": "pgrep -f 'hermes gateway' > /dev/null 2>/dev/null",
"restart": "cd /root/wizards/bezalel/hermes/venv/bin/activate && nohup hermes gateway run > /dev/null 2>&1 &",
"critical": True,
},
"evennia": {
"check": "pgrep -f 'evennia' > /dev/null 2>/dev/null",
"restart": "cd /root/.evennia/timmy_world && evennia restart 2>/dev/null",
"critical": False,
},
},
},
}
TELEGRAM_TOKEN_FILE = Path(os.path.expanduser("~/.config/telegram/special_bot"))
TELEGRAM_CHAT = "-1003664764329"
def send_telegram(message):
if not TELEGRAM_TOKEN_FILE.exists():
return False
token = TELEGRAM_TOKEN_FILE.read_text().strip()
url = f"https://api.telegram.org/bot{token}/sendMessage"
body = json.dumps({
"chat_id": TELEGRAM_CHAT,
"text": f"[AUTO-RESTART]\n{message}",
}).encode()
try:
import urllib.request
req = urllib.request.Request(url, data=body, headers={"Content-Type": "application/json"}, method="POST")
urllib.request.urlopen(req, timeout=10)
return True
except Exception:
return False
def get_cooldowns():
if COOLDOWN_FILE.exists():
try:
return json.loads(COOLDOWN_FILE.read_text())
except json.JSONDecodeError:
pass
return {}
def save_cooldowns(data):
COOLDOWN_FILE.write_text(json.dumps(data, indent=2))
def check_service(check_cmd, timeout=10):
try:
proc = subprocess.run(check_cmd, shell=True, capture_output=True, timeout=timeout)
return proc.returncode == 0
except (subprocess.TimeoutExpired, subprocess.SubprocessError):
return False
def restart_service(restart_cmd, timeout=30):
try:
proc = subprocess.run(restart_cmd, shell=True, capture_output=True, timeout=timeout)
return proc.returncode == 0
except (subprocess.TimeoutExpired, subprocess.SubprocessError) as e:
return False
def try_restart_via_ssh(name, host_config, service_name):
ip = host_config["ip"]
user = host_config["user"]
service = host_config["services"][service_name]
restart_cmd = f'ssh -o StrictHostKeyChecking=no -o ConnectTimeout=10 {user}@{ip} "{service["restart"]}"'
return restart_service(restart_cmd, timeout=30)
def log_restart(service_name, machine, attempt, success):
ts = datetime.now(timezone.utc).isoformat()
status = "SUCCESS" if success else "FAILED"
log_entry = f"{ts} [{status}] {machine}/{service_name} (attempt {attempt})\n"
RESTART_LOG.parent.mkdir(parents=True, exist_ok=True)
with open(RESTART_LOG, "a") as f:
f.write(log_entry)
print(f" [{status}] {machine}/{service_name} - attempt {attempt}")
def check_and_restart():
"""Run all restart checks."""
results = []
cooldowns = get_cooldowns()
now = time.time()
# Check local services
for name, service in LOCAL_SERVICES.items():
if not check_service(service["check"]):
cooldown_key = f"local/{name}"
retries = cooldowns.get(cooldown_key, {"count": 0, "last": 0}).get("count", 0)
if retries >= MAX_RETRIES:
last = cooldowns.get(cooldown_key, {}).get("last", 0)
if now - last < COOLDOWN_PERIOD and service["critical"]:
send_telegram(f"CRITICAL: local/{name} failed {MAX_RETRIES} restart attempts. Needs human intervention.")
cooldowns[cooldown_key] = {"count": 0, "last": now}
save_cooldowns(cooldowns)
continue
success = restart_service(service["restart"])
log_restart(name, "local", retries + 1, success)
cooldowns[cooldown_key] = {"count": retries + 1 if not success else 0, "last": now}
save_cooldowns(cooldowns)
if success:
# Verify it actually started
time.sleep(3)
if check_service(service["check"]):
print(f" VERIFIED: local/{name} is running")
else:
print(f" WARNING: local/{name} restart command returned success but process not detected")
# Check VPS services
for host, host_config in VPS_SERVICES.items():
for service_name, service in host_config["services"].items():
check_cmd = f'ssh -o StrictHostKeyChecking=no -o ConnectTimeout=5 {host_config["user"]}@{host_config["ip"]} "{service["check"]}"'
if not check_service(check_cmd):
cooldown_key = f"{host}/{service_name}"
retries = cooldowns.get(cooldown_key, {"count": 0, "last": 0}).get("count", 0)
if retries >= MAX_RETRIES:
last = cooldowns.get(cooldown_key, {}).get("last", 0)
if now - last < COOLDOWN_PERIOD and service["critical"]:
send_telegram(f"CRITICAL: {host}/{service_name} failed {MAX_RETRIES} restart attempts. Needs human intervention.")
cooldowns[cooldown_key] = {"count": 0, "last": now}
save_cooldowns(cooldowns)
continue
success = try_restart_via_ssh(host, host_config, service_name)
log_restart(service_name, host, retries + 1, success)
cooldowns[cooldown_key] = {"count": retries + 1 if not success else 0, "last": now}
save_cooldowns(cooldowns)
return results
def daemon_mode():
"""Run continuously every 60 seconds."""
print("Auto-restart agent running in daemon mode (60s interval)")
print(f"Monitoring {len(LOCAL_SERVICES)} local + {sum(len(h['services']) for h in VPS_SERVICES.values())} remote services")
print(f"Max retries per cycle: {MAX_RETRIES}")
print(f"Cooldown after max retries: {COOLDOWN_PERIOD}s")
while True:
check_and_restart()
time.sleep(60)
def show_status():
"""Show restart history and cooldowns."""
cooldowns = get_cooldowns()
print("=== Restart Cooldowns ===")
for key, data in sorted(cooldowns.items()):
count = data.get("count", 0)
if count > 0:
print(f" {key}: {count} failures, last at {datetime.fromtimestamp(data.get('last',0), tz=timezone.utc).strftime('%H:%M')}")
print("\n=== Restart Log (last 20) ===")
if RESTART_LOG.exists():
lines = RESTART_LOG.read_text().strip().split("\n")
for line in lines[-20:]:
print(f" {line}")
else:
print(" No restarts logged yet.")
if __name__ == "__main__":
LOG_DIR.mkdir(parents=True, exist_ok=True)
if len(sys.argv) > 1 and sys.argv[1] == "--daemon":
daemon_mode()
elif len(sys.argv) > 1 and sys.argv[1] == "--status":
show_status()
else:
check_and_restart()

191
fleet/capacity-inventory.md Normal file
View File

@@ -0,0 +1,191 @@
# Capacity Inventory - Fleet Resource Baseline
**Last audited:** 2026-04-07 16:00 UTC
**Auditor:** Timmy (direct inspection)
---
## Fleet Resources (Paperclips Model)
Three primary resources govern the fleet:
| Resource | Role | Generation | Consumption |
|----------|------|-----------|-------------|
| **Capacity** | Compute hours available across fleet. Determines what work can be done. | Through healthy utilization of VPS/Mac agents | Fleet improvements consume it (investing in automation, orchestration, sovereignty) |
| **Uptime** | % time services are running. Earned at Fibonacci milestones. | When services stay up naturally | Degrades on any failure |
| **Innovation** | Only generates when capacity is <70% utilized. Fuels Phase 3+. | When you leave capacity free | Phase 3+ buildings consume it (requires spare capacity to build) |
### The Tension
- Run fleet at 95%+ capacity: maximum productivity, ZERO Innovation
- Run fleet at <70% capacity: Innovation generates but slower progress
- This forces the Paperclips question: optimize now or invest in future capability?
---
## VPS Resource Baselines
### Ezra (143.198.27.163) - "Forge"
| Metric | Value | Utilization |
|--------|-------|-------------|
| **OS** | Ubuntu 24.04 (6.8.0-106-generic) | |
| **vCPU** | 4 vCPU (DO basic droplet, shared) | Load: 10.76/7.59/7.04 (very high) |
| **RAM** | 7,941 MB total | 2,104 used / 5,836 available (26% used, 74% free) |
| **Disk** | 154 GB vda1 | 111 GB used / 44 GB free (72%) **WARNING** |
| **Swap** | 6,143 MB | 643 MB used (10%) |
| **Uptime** | 7 days, 18 hours | |
### Key Processes (sorted by memory)
| Process | RSS | %CPU | Notes |
|---------|-----|------|-------|
| Gitea | 556 MB | 83.5% | Web service, high CPU due to API load |
| MemPalace (ezra) | 268 MB | 136% | Mining project files - HIGH CPU |
| Hermes gateway (ezra) | 245 MB | 1.7% | Agent gateway |
| Ollama | 230 MB | 0.1% | Model serving |
| PostgreSQL | 138 MB | ~0% | Gitea database |
**Capacity assessment:** 26% memory used, but 72% disk is getting tight. CPU load is very high (10.76 on 4vCPU = 269% utilization). Ezra is CPU-bound, not RAM-bound.
### Allegro (167.99.126.228)
| Metric | Value | Utilization |
|--------|-------|-------------|
| **OS** | Ubuntu 24.04 (6.8.0-106-generic) | |
| **vCPU** | 4 vCPU (DO basic droplet, shared) | Moderate load |
| **RAM** | 7,941 MB total | 1,591 used / 6,349 available (20% used, 80% free) |
| **Disk** | 154 GB vda1 | 41 GB used / 114 GB free (27%) **GOOD** |
| **Swap** | 8,191 MB | 686 MB used (8%) |
| **Uptime** | 7 days, 18 hours | |
### Key Processes (sorted by memory)
| Process | RSS | %CPU | Notes |
|---------|-----|------|-------|
| Hermes gateway (allegro) | 680 MB | 0.9% | Main agent gateway |
| Gitea | 181 MB | 1.2% | Secondary gitea? |
| Systemd-journald | 160 MB | 0.0% | System logging |
| Ezra Hermes gateway | 58 MB | 0.0% | Running ezra agent here |
| Bezalel Hermes gateway | 58 MB | 0.0% | Running bezalel agent here |
| Dockerd | 48 MB | 0.0% | Docker daemon |
**Capacity assessment:** 20% memory used, 27% disk used. Allegro has headroom. Also running hermes gateways for Ezra and Bezalel (cross-host agent execution).
### Bezalel (159.203.146.185)
| Metric | Value | Utilization |
|--------|-------|-------------|
| **OS** | Ubuntu 24.04 (6.8.0-71-generic) | |
| **vCPU** | 2 vCPU (DO basic droplet, shared) | Load varies |
| **RAM** | 1,968 MB total | 817 used / 1,151 available (42% used, 58% free) |
| **Disk** | 48 GB vda1 | 12 GB used / 37 GB free (24%) **GOOD** |
| **Swap** | 2,047 MB | 448 MB used (22%) |
| **Uptime** | 7 days, 18 hours | |
### Key Processes (sorted by memory)
| Process | RSS | %CPU | Notes |
|---------|-----|------|-------|
| Hermes gateway | 339 MB | 7.7% | Agent gateway (16.8% of RAM) |
| uv pip install | 137 MB | 56.6% | Installing packages (temporary) |
| Mender | 27 MB | 0.0% | Device management |
**Capacity assessment:** 42% memory used, only 2GB total RAM. Bezalel is the most constrained. 2 vCPU means less compute headroom than Ezra/Allegro. Disk is fine.
### Mac Local (M3 Max)
| Metric | Value | Utilization |
|--------|-------|-------------|
| **OS** | macOS 26.3.1 | |
| **CPU** | Apple M3 Max (14 cores) | Very capable |
| **RAM** | 36 GB | ~8 GB used (22%) |
| **Disk** | 926 GB total | ~624 GB used / 302 GB free (68%) |
### Key Processes
| Process | Memory | Notes |
|---------|--------|-------|
| Hermes gateway | 500 MB | Primary gateway |
| Hermes agents (x3) | ~560 MB total | Multiple sessions |
| Ollama | ~20 MB base + model memory | Model loading varies |
| OpenClaw | 350 MB | Gateway process |
| Evennia (server+portal) | 56 MB | Game world |
---
## Resource Summary
| Resource | Ezra | Allegro | Bezalel | Mac Local | TOTAL |
|----------|------|---------|---------|-----------|-------|
| **vCPU** | 4 | 4 | 2 | 14 (M3 Max) | 24 |
| **RAM** | 8 GB (26% used) | 8 GB (20% used) | 2 GB (42% used) | 36 GB (22% used) | 54 GB |
| **Disk** | 154 GB (72%) | 154 GB (27%) | 48 GB (24%) | 926 GB (68%) | 1,282 GB |
| **Cost** | $12/mo | $12/mo | $12/mo | owned | $36/mo |
### Utilization by Category
| Category | Estimated Daily Hours | % of Fleet Capacity |
|----------|----------------------|---------------------|
| Hermes agents | ~3-4 hrs active | 5-7% |
| Ollama inference | ~1-2 hrs | 2-4% |
| Gitea services | 24/7 | 5-10% |
| Evennia | 24/7 | <1% |
| Idle | ~18-20 hrs | ~80-90% |
### Capacity Utilization: ~15-20% active
**Innovation rate:** GENERATING (capacity < 70%)
**Recommendation:** Good — Innovation is generating because most capacity is free.
This means Phase 3+ capabilities (orchestration, load balancing, etc.) are accessible NOW.
---
## Uptime Baseline
**Baseline period:** 2026-04-07 14:00-16:00 UTC (2 hours, ~24 checks at 5-min intervals)
| Service | Checks | Uptime | Status |
|---------|--------|--------|--------|
| Ezra | 24/24 | 100.0% | GOOD |
| Allegro | 24/24 | 100.0% | GOOD |
| Bezalel | 24/24 | 100.0% | GOOD |
| Gitea | 23/24 | 95.8% | GOOD |
| Hermes Gateway | 23/24 | 95.8% | GOOD |
| Ollama | 24/24 | 100.0% | GOOD |
| OpenClaw | 24/24 | 100.0% | GOOD |
| Evennia | 24/24 | 100.0% | GOOD |
| Hermes Agent | 21/24 | 87.5% | **CHECK** |
### Fibonacci Uptime Milestones
| Milestone | Target | Current | Status |
|-----------|--------|---------|--------|
| 95% | 95% | 100% (VPS), 98.6% (avg) | REACHED |
| 95.5% | 95.5% | 98.6% | REACHED |
| 96% | 96% | 98.6% | REACHED |
| 97% | 97% | 98.6% | REACHED |
| 98% | 98% | 98.6% | REACHED |
| 99% | 99% | 98.6% | APPROACHING |
---
## Risk Assessment
| Risk | Severity | Mitigation |
|------|----------|------------|
| Ezra disk 72% used | MEDIUM | Move non-essential data, add monitoring alert at 85% |
| Bezalel only 2GB RAM | HIGH | Cannot run large models locally. Good for Evennia, tight for agents |
| Ezra CPU load 269% | HIGH | MemPalace mining consuming 136% CPU. Consider scheduling |
| Mac disk 68% used | MEDIUM | 302 GB free still. Growing but not urgent |
| No cross-VPS mesh | LOW | SSH works but no Tailscale. No private network between VPSes |
---
## Recommendations
### Immediate (Phase 1-2)
1. **Ezra disk cleanup:** 44 GB free at 72%. Docker images, old logs, and MemPalace mine data could be rotated.
2. **Alert thresholds:** Add disk alerts at 85% (Ezra, Mac) before they become critical.
### Short-term (Phase 3)
3. **Load balancing:** Ezra is CPU-bound, Allegro has 80% RAM free. Move some agent processes from Ezra to Allegro.
4. **Innovation investment:** Since fleet is at 15-20% utilization, Innovation is high. This is the time to build Phase 3 capabilities.
### Medium-term (Phase 4)
5. **Bezalel RAM upgrade:** 2GB is tight. Consider upgrade to 4GB ($24/mo instead of $12/mo).
6. **Tailscale mesh:** Install on all VPSes for private inter-VPS network.
---

142
fleet/milestones.md Normal file
View File

@@ -0,0 +1,142 @@
# Fleet Milestone Messages
Every milestone marks passage through fleet evolution. When achieved, the message
prints to the fleet log. Each one references a real achievement, not abstract numbers.
**Source:** Inspired by Paperclips milestone messages (500 clips, 1000 clips, Full autonomy attained, etc.)
---
## Phase 1: Survival (Current)
### M1: First Automated Health Check
**Trigger:** `fleet/health_check.py` runs successfully for the first time.
**Message:** "First automated health check runs. No longer watching the clock."
### M2: First Auto-Restart
**Trigger:** A dead process is detected and restarted without human intervention.
**Message:** "A process failed at 3am and restarted itself. You found out in the morning."
### M3: First Backup Completed
**Trigger:** A backup pipeline runs end-to-end and verifies integrity.
**Message:** "A backup completed. You did not have to think about it."
### M4: 95% Uptime (30 days)
**Trigger:** Uptime >= 95% over last 30 days.
**Message:** "95% uptime over 30 days. The fleet stays up."
### M5: Uptime 97%
**Trigger:** Uptime >= 97% over last 30 days.
**Message:** "97% uptime. Three nines of availability across four machines."
---
## Phase 2: Automation (unlock when: uptime >= 95% + capacity > 60%)
### M6: Zero Manual Restarts (7 days)
**Trigger:** 7 consecutive days with zero manual process restarts.
**Message:** "Seven days. Zero manual restarts. The fleet heals itself."
### M7: PR Auto-Merged
**Trigger:** A PR passes CI, review, and merges without human touching it.
**Message:** "A PR was tested, reviewed, and merged by agents. You just said 'looks good.'"
### M8: Config Push Works
**Trigger:** Config change pushed to all 3 VPSes atomically and verified.
**Message:** "Config pushed to all three VPSes in one command. No SSH needed."
### M9: 98% Uptime
**Trigger:** Uptime >= 98% over last 30 days.
**Message:** "98% uptime. Only 14 hours of downtime in a month. Most of it planned."
---
## Phase 3: Orchestration (unlock when: all Phase 2 buildings + Innovation > 100)
### M10: Cross-Agent Delegation Works
**Trigger:** Agent A creates issue, assigns to Agent B, Agent B works and creates PR.
**Message:** "Agent Alpha created a task, Agent Beta completed it. They did not ask permission."
### M11: First Model Running Locally on 2+ Machines
**Trigger:** Ollama serving same model on Ezra and Allegro simultaneously.
**Message:** "A model runs on two machines at once. No cloud. No rate limits."
### M12: Fleet-Wide Burn Mode
**Trigger:** All agents coordinated on single epic, produced coordinated PRs.
**Message:** "All agents working the same epic. The fleet moves as one."
---
## Phase 4: Sovereignty (unlock when: zero cloud deps for core ops)
### M13: First Entirely Local Inference Day
**Trigger:** 24 hours with zero API calls to external providers.
**Message:** "A model ran locally for the first time. No cloud. No rate limits. No one can turn it off."
### M14: Sovereign Email
**Trigger:** Stalwart email server sends and receives without Gmail relay.
**Message:** "Email flows through our own server. No Google. No Microsoft. Ours."
### M15: Sovereign Messaging
**Trigger:** Telegram bot runs without cloud relay dependency.
**Message:** "Messages arrive through our own infrastructure. No corporate middleman."
---
## Phase 5: Scale (unlock when: sovereignty stable + Innovation > 500)
### M16: First Self-Spawned Agent
**Trigger:** Agent lifecycle manager spawns a new agent instance due to load.
**Message:** "A new agent appeared. You did not create it. The fleet built what it needed."
### M17: Agent Retired Gracefully
**Trigger:** An agent instance retires after idle timeout and cleans up its state.
**Message:** "An agent retired. It served its purpose. Nothing was lost."
### M18: Fleet Runs 24h Unattended
**Trigger:** 24 hours with zero human intervention of any kind.
**Message:** "A full day. No humans. No commands. The fleet runs itself."
---
## Phase 6: The Network (unlock when: 7 days zero human intervention)
### M19: Fleet Creates Its Own Improvement Task
**Trigger:** Fleet analyzes itself and creates an issue on Gitea.
**Message:** "The fleet found something to improve. It created the task itself."
### M20: First Outside Contribution
**Trigger:** An external contributor's PR is reviewed and merged by fleet agents.
**Message:** "Someone outside the fleet contributed. The fleet reviewed, tested, and merged. No human touched it."
### M21: The Beacon
**Trigger:** Infrastructure serves someone in need through automated systems.
**Message:** "Someone found the Beacon. In the dark, looking for help. The infrastructure served its purpose. It was built for this."
### M22: Permanent Light
**Trigger:** 90 days of autonomous operation with continuous availability.
**Message:** "Three months. The light never went out. Not for anyone."
---
## Fibonacci Uptime Milestones
These trigger regardless of phase, based purely on uptime percentage:
| Milestone | Uptime | Meaning |
|-----------|--------|--------|
| U1 | 95% | Basic reliability achieved |
| U2 | 95.5% | Fewer than 16 hours/month downtime |
| U3 | 96% | Fewer than 12 hours/month |
| U4 | 97% | Fewer than 9 hours/month |
| U5 | 97.5% | Fewer than 7 hours/month |
| U6 | 98% | Fewer than 4.5 hours/month |
| U7 | 98.3% | Fewer than 3 hours/month |
| U8 | 98.6% | Less than 2.5 hours/month — approaching cloud tier |
| U9 | 98.9% | Less than 1.5 hours/month |
| U10 | 99% | Less than 1 hour/month — enterprise grade |
| U11 | 99.5% | Less than 22 minutes/month |
---
*Every message is earned. None are given freely. Fleet evolution is not a checklist — it is a climb.*

231
fleet/resource_tracker.py Executable file
View File

@@ -0,0 +1,231 @@
#!/usr/bin/env python3
"""
Fleet Resource Tracker — Tracks Capacity, Uptime, and Innovation.
Paperclips-inspired tension model:
- Capacity: spent on fleet improvements, generates through utilization
- Uptime: earned when services stay up, Fibonacci milestones unlock capabilities
- Innovation: only generates when capacity < 70%. Fuels Phase 3+.
This is the heart of the fleet progression system.
"""
import os
import json
import time
import socket
from datetime import datetime, timezone
from pathlib import Path
# === CONFIG ===
DATA_DIR = Path(os.path.expanduser("~/.local/timmy/fleet-resources"))
RESOURCES_FILE = DATA_DIR / "resources.json"
# Tension thresholds
INNOVATION_THRESHOLD = 0.70 # Innovation only generates when capacity < 70%
INNOVATION_RATE = 5.0 # Innovation generated per hour when under threshold
CAPACITY_REGEN_RATE = 2.0 # Capacity regenerates per hour of healthy operation
FIBONACCI = [95.0, 95.5, 96.0, 97.0, 97.5, 98.0, 98.3, 98.6, 98.9, 99.0, 99.5]
def init():
DATA_DIR.mkdir(parents=True, exist_ok=True)
if not RESOURCES_FILE.exists():
data = {
"capacity": {
"current": 100.0,
"max": 100.0,
"spent_on": [],
"history": []
},
"uptime": {
"current_pct": 100.0,
"milestones_reached": [],
"total_checks": 0,
"successful_checks": 0,
"history": []
},
"innovation": {
"current": 0.0,
"total_generated": 0.0,
"spent_on": [],
"last_calculated": time.time()
}
}
RESOURCES_FILE.write_text(json.dumps(data, indent=2))
print("Initialized resource tracker")
return RESOURCES_FILE.exists()
def load():
if RESOURCES_FILE.exists():
return json.loads(RESOURCES_FILE.read_text())
return None
def save(data):
RESOURCES_FILE.write_text(json.dumps(data, indent=2))
def update_uptime(checks: dict):
"""Update uptime stats from health check results.
checks = {'ezra': True, 'allegro': True, 'bezalel': True, 'gitea': True, ...}
"""
data = load()
if not data:
return
data["uptime"]["total_checks"] += 1
successes = sum(1 for v in checks.values() if v)
total = len(checks)
# Overall uptime percentage
overall = successes / max(total, 1) * 100.0
data["uptime"]["successful_checks"] += successes
# Calculate rolling uptime
if "history" not in data["uptime"]:
data["uptime"]["history"] = []
data["uptime"]["history"].append({
"ts": datetime.now(timezone.utc).isoformat(),
"checks": checks,
"overall": round(overall, 2)
})
# Keep last 1000 checks
if len(data["uptime"]["history"]) > 1000:
data["uptime"]["history"] = data["uptime"]["history"][-1000:]
# Calculate current uptime %, last 100 checks
recent = data["uptime"]["history"][-100:]
recent_ok = sum(c["overall"] for c in recent) / max(len(recent), 1)
data["uptime"]["current_pct"] = round(recent_ok, 2)
# Check Fibonacci milestones
new_milestones = []
for fib in FIBONACCI:
if fib not in data["uptime"]["milestones_reached"] and recent_ok >= fib:
data["uptime"]["milestones_reached"].append(fib)
new_milestones.append(fib)
save(data)
if new_milestones:
print(f" UPTIME MILESTONE: {','.join(str(m) + '%') for m in new_milestones}")
print(f" Current uptime: {recent_ok:.1f}%")
return data["uptime"]
def spend_capacity(amount: float, purpose: str):
"""Spend capacity on a fleet improvement."""
data = load()
if not data:
return False
if data["capacity"]["current"] < amount:
print(f" INSUFFICIENT CAPACITY: Need {amount}, have {data['capacity']['current']:.1f}")
return False
data["capacity"]["current"] -= amount
data["capacity"]["spent_on"].append({
"purpose": purpose,
"amount": amount,
"ts": datetime.now(timezone.utc).isoformat()
})
save(data)
print(f" Spent {amount} capacity on: {purpose}")
return True
def regenerate_resources():
"""Regenerate capacity and calculate innovation."""
data = load()
if not data:
return
now = time.time()
last = data["innovation"]["last_calculated"]
hours = (now - last) / 3600.0
if hours < 0.1: # Only update every ~6 minutes
return
# Regenerate capacity
capacity_gain = CAPACITY_REGEN_RATE * hours
data["capacity"]["current"] = min(
data["capacity"]["max"],
data["capacity"]["current"] + capacity_gain
)
# Calculate capacity utilization
utilization = 1.0 - (data["capacity"]["current"] / data["capacity"]["max"])
# Generate innovation only when under threshold
innovation_gain = 0.0
if utilization < INNOVATION_THRESHOLD:
innovation_gain = INNOVATION_RATE * hours * (1.0 - utilization / INNOVATION_THRESHOLD)
data["innovation"]["current"] += innovation_gain
data["innovation"]["total_generated"] += innovation_gain
# Record history
if "history" not in data["capacity"]:
data["capacity"]["history"] = []
data["capacity"]["history"].append({
"ts": datetime.now(timezone.utc).isoformat(),
"capacity": round(data["capacity"]["current"], 1),
"utilization": round(utilization * 100, 1),
"innovation": round(data["innovation"]["current"], 1),
"innovation_gain": round(innovation_gain, 1)
})
# Keep last 500 capacity records
if len(data["capacity"]["history"]) > 500:
data["capacity"]["history"] = data["capacity"]["history"][-500:]
data["innovation"]["last_calculated"] = now
save(data)
print(f" Capacity: {data['capacity']['current']:.1f}/{data['capacity']['max']:.1f}")
print(f" Utilization: {utilization*100:.1f}%")
print(f" Innovation: {data['innovation']['current']:.1f} (+{innovation_gain:.1f} this period)")
return data
def status():
"""Print current resource status."""
data = load()
if not data:
print("Resource tracker not initialized. Run --init first.")
return
print("\n=== Fleet Resources ===")
print(f" Capacity: {data['capacity']['current']:.1f}/{data['capacity']['max']:.1f}")
utilization = 1.0 - (data["capacity"]["current"] / data["capacity"]["max"])
print(f" Utilization: {utilization*100:.1f}%")
innovation_status = "GENERATING" if utilization < INNOVATION_THRESHOLD else "BLOCKED"
print(f" Innovation: {data['innovation']['current']:.1f} [{innovation_status}]")
print(f" Uptime: {data['uptime']['current_pct']:.1f}%")
print(f" Milestones: {', '.join(str(m)+'%' for m in data['uptime']['milestones_reached']) or 'None yet'}")
# Phase gate checks
phase_2_ok = data['uptime']['current_pct'] >= 95.0
phase_3_ok = phase_2_ok and data['innovation']['current'] > 100
phase_5_ok = phase_2_ok and data['innovation']['current'] > 500
print(f"\n Phase Gates:")
print(f" Phase 2 (Automation): {'UNLOCKED' if phase_2_ok else 'LOCKED (need 95% uptime)'}")
print(f" Phase 3 (Orchestration): {'UNLOCKED' if phase_3_ok else 'LOCKED (need 95% uptime + 100 innovation)'}")
print(f" Phase 5 (Scale): {'UNLOCKED' if phase_5_ok else 'LOCKED (need 95% uptime + 500 innovation)'}")
if __name__ == "__main__":
import sys
init()
if len(sys.argv) > 1 and sys.argv[1] == "status":
status()
elif len(sys.argv) > 1 and sys.argv[1] == "regen":
regenerate_resources()
else:
regenerate_resources()
status()

View File

@@ -146,6 +146,7 @@ class PullRequest:
additions: int = 0
deletions: int = 0
created_at: str = ""
updated_at: str = ""
closed_at: str = ""
@classmethod
@@ -166,6 +167,7 @@ class PullRequest:
additions=d.get("additions", 0),
deletions=d.get("deletions", 0),
created_at=d.get("created_at", ""),
updated_at=d.get("updated_at", ""),
closed_at=d.get("closed_at", ""),
)
@@ -314,6 +316,7 @@ class GiteaClient:
direction: str = "desc",
limit: int = 30,
page: int = 1,
since: Optional[str] = None,
) -> list[Issue]:
"""List issues for a repo."""
raw = self._get(
@@ -326,6 +329,7 @@ class GiteaClient:
direction=direction,
limit=limit,
page=page,
since=since,
)
return [Issue.from_dict(i) for i in raw]

View File

@@ -0,0 +1,39 @@
#!/usr/bin/env bash
# orchestrate.sh — Sovereign Orchestrator wrapper
# Sets environment and runs orchestrator.py
#
# Usage:
# ./orchestrate.sh # dry-run (safe default)
# ./orchestrate.sh --once # single live dispatch cycle
# ./orchestrate.sh --daemon # continuous (every 15 min)
# ./orchestrate.sh --dry-run # explicit dry-run
set -euo pipefail
SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
HERMES_DIR="${HOME}/.hermes"
# Load Gitea token
if [[ -z "${GITEA_TOKEN:-}" ]]; then
if [[ -f "${HERMES_DIR}/gitea_token_vps" ]]; then
export GITEA_TOKEN="$(cat "${HERMES_DIR}/gitea_token_vps")"
else
echo "[FATAL] No GITEA_TOKEN and ~/.hermes/gitea_token_vps not found"
exit 1
fi
fi
# Load Telegram token
if [[ -z "${TELEGRAM_BOT_TOKEN:-}" ]]; then
if [[ -f "${HOME}/.config/telegram/special_bot" ]]; then
export TELEGRAM_BOT_TOKEN="$(cat "${HOME}/.config/telegram/special_bot")"
fi
fi
# Run preflight checks if available
if [[ -x "${HERMES_DIR}/bin/api-key-preflight.sh" ]]; then
"${HERMES_DIR}/bin/api-key-preflight.sh" 2>/dev/null || true
fi
# Run the orchestrator
exec python3 "${SCRIPT_DIR}/orchestrator.py" "$@"

View File

@@ -0,0 +1,645 @@
#!/usr/bin/env python3
"""
Sovereign Orchestrator v1
Reads the Gitea backlog, scores/prioritizes issues, dispatches to agents.
Usage:
python3 orchestrator.py --once # single dispatch cycle
python3 orchestrator.py --daemon # run every 15 min
python3 orchestrator.py --dry-run # score and report, no dispatch
"""
import json
import os
import sys
import time
import subprocess
import urllib.request
import urllib.error
import urllib.parse
from datetime import datetime, timezone
# ---------------------------------------------------------------------------
# CONFIG
# ---------------------------------------------------------------------------
GITEA_API = "https://forge.alexanderwhitestone.com/api/v1"
GITEA_OWNER = "Timmy_Foundation"
REPOS = ["timmy-config", "the-nexus", "timmy-home"]
TELEGRAM_CHAT_ID = "-1003664764329"
DAEMON_INTERVAL = 900 # 15 minutes
# Tags that mark issues we should never auto-dispatch
FILTER_TAGS = ["[EPIC]", "[DO NOT CLOSE]", "[PERMANENT]", "[PHILOSOPHY]", "[MORNING REPORT]"]
# Known agent usernames on Gitea (for assignee detection)
AGENT_USERNAMES = {"groq", "ezra", "bezalel", "allegro", "timmy", "thetimmyc"}
# ---------------------------------------------------------------------------
# AGENT ROSTER
# ---------------------------------------------------------------------------
AGENTS = {
"groq": {
"type": "loop",
"endpoint": "local",
"strengths": ["code", "bug-fix", "small-changes"],
"repos": ["the-nexus", "hermes-agent", "timmy-config", "timmy-home"],
"max_concurrent": 1,
},
"ezra": {
"type": "gateway",
"endpoint": "http://143.198.27.163:8643/v1/chat/completions",
"ssh": "root@143.198.27.163",
"strengths": ["research", "architecture", "complex", "multi-file"],
"repos": ["timmy-config", "the-nexus", "timmy-home"],
"max_concurrent": 1,
},
"bezalel": {
"type": "gateway",
"endpoint": "http://159.203.146.185:8643/v1/chat/completions",
"ssh": "root@159.203.146.185",
"strengths": ["ci", "infra", "ops", "testing"],
"repos": ["timmy-config", "hermes-agent", "the-nexus"],
"max_concurrent": 1,
},
}
# ---------------------------------------------------------------------------
# CREDENTIALS
# ---------------------------------------------------------------------------
def load_gitea_token():
"""Read Gitea token from env or file."""
token = os.environ.get("GITEA_TOKEN", "")
if token:
return token.strip()
token_path = os.path.expanduser("~/.hermes/gitea_token_vps")
try:
with open(token_path) as f:
return f.read().strip()
except FileNotFoundError:
print(f"[FATAL] No GITEA_TOKEN env and {token_path} not found")
sys.exit(1)
def load_telegram_token():
"""Read Telegram bot token from file."""
path = os.path.expanduser("~/.config/telegram/special_bot")
try:
with open(path) as f:
return f.read().strip()
except FileNotFoundError:
return ""
GITEA_TOKEN = ""
TELEGRAM_TOKEN = ""
# ---------------------------------------------------------------------------
# HTTP HELPERS (stdlib only)
# ---------------------------------------------------------------------------
def gitea_request(path, method="GET", data=None):
"""Make an authenticated Gitea API request."""
url = f"{GITEA_API}{path}"
headers = {
"Authorization": f"token {GITEA_TOKEN}",
"Content-Type": "application/json",
"Accept": "application/json",
}
body = json.dumps(data).encode() if data else None
req = urllib.request.Request(url, data=body, headers=headers, method=method)
try:
with urllib.request.urlopen(req, timeout=30) as resp:
return json.loads(resp.read().decode())
except urllib.error.HTTPError as e:
body_text = e.read().decode() if e.fp else ""
print(f"[API ERROR] {method} {url} -> {e.code}: {body_text[:200]}")
return None
except Exception as e:
print(f"[API ERROR] {method} {url} -> {e}")
return None
def send_telegram(message):
"""Send message to Telegram group."""
if not TELEGRAM_TOKEN:
print("[WARN] No Telegram token, skipping notification")
return False
url = f"https://api.telegram.org/bot{TELEGRAM_TOKEN}/sendMessage"
data = json.dumps({
"chat_id": TELEGRAM_CHAT_ID,
"text": message,
"parse_mode": "Markdown",
"disable_web_page_preview": True,
}).encode()
req = urllib.request.Request(url, data=data, headers={"Content-Type": "application/json"})
try:
with urllib.request.urlopen(req, timeout=15) as resp:
return resp.status == 200
except Exception as e:
print(f"[TELEGRAM ERROR] {e}")
return False
# ---------------------------------------------------------------------------
# 1. BACKLOG READER
# ---------------------------------------------------------------------------
def fetch_issues(repo):
"""Fetch all open issues from a repo, handling pagination."""
issues = []
page = 1
while True:
result = gitea_request(
f"/repos/{GITEA_OWNER}/{repo}/issues?state=open&type=issues&limit=50&page={page}"
)
if not result:
break
issues.extend(result)
if len(result) < 50:
break
page += 1
return issues
def should_filter(issue):
"""Check if issue title contains any filter tags."""
title = issue.get("title", "").upper()
for tag in FILTER_TAGS:
if tag.upper().replace("[", "").replace("]", "") in title.replace("[", "").replace("]", ""):
return True
# Also filter pull requests
if issue.get("pull_request"):
return True
return False
def read_backlog():
"""Read and filter the full backlog across all repos."""
backlog = []
for repo in REPOS:
print(f" Fetching {repo}...")
issues = fetch_issues(repo)
for issue in issues:
if should_filter(issue):
continue
assignees = [a.get("login", "") for a in (issue.get("assignees") or [])]
labels = [l.get("name", "") for l in (issue.get("labels") or [])]
backlog.append({
"repo": repo,
"number": issue["number"],
"title": issue["title"],
"labels": labels,
"assignees": assignees,
"created_at": issue.get("created_at", ""),
"comments": issue.get("comments", 0),
"url": issue.get("html_url", ""),
})
print(f" Total actionable issues: {len(backlog)}")
return backlog
# ---------------------------------------------------------------------------
# 2. PRIORITY SCORER
# ---------------------------------------------------------------------------
def score_issue(issue):
"""Score an issue 0-100 based on priority signals."""
score = 0
title_upper = issue["title"].upper()
labels_upper = [l.upper() for l in issue["labels"]]
all_text = title_upper + " " + " ".join(labels_upper)
# Critical / Bug: +30
if any(tag in all_text for tag in ["CRITICAL", "BUG"]):
score += 30
# P0 / Urgent: +25
if any(tag in all_text for tag in ["P0", "URGENT"]):
score += 25
# P1: +15
if "P1" in all_text:
score += 15
# OPS / Security: +10
if any(tag in all_text for tag in ["OPS", "SECURITY"]):
score += 10
# Unassigned: +10
if not issue["assignees"]:
score += 10
# Age > 7 days: +5
try:
created = issue["created_at"].replace("Z", "+00:00")
created_dt = datetime.fromisoformat(created)
age_days = (datetime.now(timezone.utc) - created_dt).days
if age_days > 7:
score += 5
except (ValueError, AttributeError):
pass
# Has comments: +5
if issue["comments"] > 0:
score += 5
# Infrastructure repo: +5
if issue["repo"] == "timmy-config":
score += 5
# Already assigned to an agent: -10
if any(a.lower() in AGENT_USERNAMES for a in issue["assignees"]):
score -= 10
issue["score"] = max(0, min(100, score))
return issue
def prioritize_backlog(backlog):
"""Score and sort the backlog by priority."""
scored = [score_issue(i) for i in backlog]
scored.sort(key=lambda x: x["score"], reverse=True)
return scored
# ---------------------------------------------------------------------------
# 3. AGENT HEALTH CHECKS
# ---------------------------------------------------------------------------
def check_process(pattern):
"""Check if a local process matching pattern is running."""
try:
result = subprocess.run(
["pgrep", "-f", pattern],
capture_output=True, text=True, timeout=5
)
return result.returncode == 0
except Exception:
return False
def check_ssh_service(host, service_name):
"""Check if a remote service is running via SSH."""
try:
result = subprocess.run(
["ssh", "-o", "ConnectTimeout=5", "-o", "StrictHostKeyChecking=no",
f"root@{host}",
f"systemctl is-active {service_name} 2>/dev/null || pgrep -f {service_name}"],
capture_output=True, text=True, timeout=15
)
return result.returncode == 0
except Exception:
return False
def check_agent_health(name, agent):
"""Check if an agent is alive and available."""
if agent["type"] == "loop":
alive = check_process(f"agent-loop.*{name}")
elif agent["type"] == "gateway":
host = agent["ssh"].split("@")[1]
service = f"hermes-{name}"
alive = check_ssh_service(host, service)
else:
alive = False
return alive
def get_agent_status():
"""Get health status for all agents."""
status = {}
for name, agent in AGENTS.items():
alive = check_agent_health(name, agent)
status[name] = {
"alive": alive,
"type": agent["type"],
"strengths": agent["strengths"],
}
symbol = "UP" if alive else "DOWN"
print(f" {name}: {symbol} ({agent['type']})")
return status
# ---------------------------------------------------------------------------
# 4. DISPATCHER
# ---------------------------------------------------------------------------
def classify_issue(issue):
"""Classify issue type based on title and labels."""
title = issue["title"].upper()
labels = " ".join(issue["labels"]).upper()
all_text = title + " " + labels
types = []
if any(w in all_text for w in ["BUG", "FIX", "BROKEN", "ERROR", "CRASH"]):
types.append("bug-fix")
if any(w in all_text for w in ["OPS", "DEPLOY", "CI", "INFRA", "PIPELINE", "MONITOR"]):
types.append("ops")
if any(w in all_text for w in ["SECURITY", "AUTH", "TOKEN", "CERT"]):
types.append("ops")
if any(w in all_text for w in ["RESEARCH", "AUDIT", "INVESTIGATE", "EXPLORE"]):
types.append("research")
if any(w in all_text for w in ["ARCHITECT", "DESIGN", "REFACTOR", "REWRITE"]):
types.append("architecture")
if any(w in all_text for w in ["TEST", "TESTING", "QA", "VALIDATE"]):
types.append("testing")
if any(w in all_text for w in ["CODE", "IMPLEMENT", "ADD", "CREATE", "BUILD"]):
types.append("code")
if any(w in all_text for w in ["SMALL", "QUICK", "SIMPLE", "MINOR", "TWEAK"]):
types.append("small-changes")
if any(w in all_text for w in ["COMPLEX", "MULTI", "LARGE", "OVERHAUL"]):
types.append("complex")
if not types:
types = ["code"] # default
return types
def match_agent(issue, agent_status, dispatched_this_cycle):
"""Find the best available agent for an issue."""
issue_types = classify_issue(issue)
candidates = []
for name, agent in AGENTS.items():
# Agent must be alive
if not agent_status.get(name, {}).get("alive", False):
continue
# Agent must handle this repo
if issue["repo"] not in agent["repos"]:
continue
# Agent must not already be dispatched this cycle
if dispatched_this_cycle.get(name, 0) >= agent["max_concurrent"]:
continue
# Score match based on overlapping strengths
overlap = len(set(issue_types) & set(agent["strengths"]))
candidates.append((name, overlap))
if not candidates:
return None
# Sort by overlap score descending, return best match
candidates.sort(key=lambda x: x[1], reverse=True)
return candidates[0][0]
def assign_issue(repo, number, agent_name):
"""Assign an issue to an agent on Gitea."""
# First get current assignees to not clobber
result = gitea_request(f"/repos/{GITEA_OWNER}/{repo}/issues/{number}")
if not result:
return False
current = [a.get("login", "") for a in (result.get("assignees") or [])]
if agent_name in current:
print(f" Already assigned to {agent_name}")
return True
new_assignees = current + [agent_name]
patch_result = gitea_request(
f"/repos/{GITEA_OWNER}/{repo}/issues/{number}",
method="PATCH",
data={"assignees": new_assignees}
)
return patch_result is not None
def dispatch_to_gateway(agent_name, agent, issue):
"""Trigger work on a gateway agent via SSH."""
host = agent["ssh"]
repo = issue["repo"]
number = issue["number"]
title = issue["title"]
# Try to trigger dispatch via SSH
cmd = (
f'ssh -o ConnectTimeout=10 -o StrictHostKeyChecking=no {host} '
f'"echo \'Dispatched by orchestrator: {repo}#{number} - {title}\' '
f'>> /tmp/hermes-dispatch.log"'
)
try:
subprocess.run(cmd, shell=True, timeout=20, capture_output=True)
return True
except Exception as e:
print(f" [WARN] SSH dispatch to {agent_name} failed: {e}")
return False
def dispatch_cycle(backlog, agent_status, dry_run=False):
"""Run one dispatch cycle. Returns dispatch report."""
dispatched = []
skipped = []
dispatched_count = {} # agent_name -> count dispatched this cycle
# Only dispatch unassigned issues (or issues not assigned to agents)
for issue in backlog:
agent_assigned = any(a.lower() in AGENT_USERNAMES for a in issue["assignees"])
if agent_assigned:
skipped.append((issue, "already assigned to agent"))
continue
if issue["score"] < 5:
skipped.append((issue, "score too low"))
continue
best_agent = match_agent(issue, agent_status, dispatched_count)
if not best_agent:
skipped.append((issue, "no available agent"))
continue
if dry_run:
dispatched.append({
"agent": best_agent,
"repo": issue["repo"],
"number": issue["number"],
"title": issue["title"],
"score": issue["score"],
"dry_run": True,
})
dispatched_count[best_agent] = dispatched_count.get(best_agent, 0) + 1
continue
# Actually dispatch
print(f" Dispatching {issue['repo']}#{issue['number']} -> {best_agent}")
success = assign_issue(issue["repo"], issue["number"], best_agent)
if success:
agent = AGENTS[best_agent]
if agent["type"] == "gateway":
dispatch_to_gateway(best_agent, agent, issue)
dispatched.append({
"agent": best_agent,
"repo": issue["repo"],
"number": issue["number"],
"title": issue["title"],
"score": issue["score"],
})
dispatched_count[best_agent] = dispatched_count.get(best_agent, 0) + 1
else:
skipped.append((issue, "assignment failed"))
return dispatched, skipped
# ---------------------------------------------------------------------------
# 5. CONSOLIDATED REPORT
# ---------------------------------------------------------------------------
def generate_report(backlog, dispatched, skipped, agent_status, dry_run=False):
"""Generate dispatch cycle report."""
now = datetime.now().strftime("%Y-%m-%d %H:%M")
mode = " [DRY RUN]" if dry_run else ""
lines = []
lines.append(f"=== Sovereign Orchestrator Report{mode} ===")
lines.append(f"Time: {now}")
lines.append(f"Total backlog: {len(backlog)} issues")
lines.append("")
# Agent health
lines.append("-- Agent Health --")
for name, info in agent_status.items():
symbol = "UP" if info["alive"] else "DOWN"
lines.append(f" {name}: {symbol} ({info['type']})")
lines.append("")
# Dispatched
lines.append(f"-- Dispatched: {len(dispatched)} --")
for d in dispatched:
dry = " (dry-run)" if d.get("dry_run") else ""
lines.append(f" [{d['score']}] {d['repo']}#{d['number']} -> {d['agent']}{dry}")
lines.append(f" {d['title'][:60]}")
lines.append("")
# Skipped (top 10)
skip_summary = {}
for issue, reason in skipped:
skip_summary[reason] = skip_summary.get(reason, 0) + 1
lines.append(f"-- Skipped: {len(skipped)} --")
for reason, count in sorted(skip_summary.items(), key=lambda x: -x[1]):
lines.append(f" {reason}: {count}")
lines.append("")
# Top 5 unassigned
unassigned = [i for i in backlog if not i["assignees"]][:5]
lines.append("-- Top 5 Unassigned (by priority) --")
for i in unassigned:
lines.append(f" [{i['score']}] {i['repo']}#{i['number']}: {i['title'][:55]}")
lines.append("")
report = "\n".join(lines)
return report
def format_telegram_report(backlog, dispatched, skipped, agent_status, dry_run=False):
"""Format a compact Telegram message."""
mode = " DRY RUN" if dry_run else ""
now = datetime.now().strftime("%H:%M")
parts = [f"*Orchestrator{mode}* ({now})"]
parts.append(f"Backlog: {len(backlog)} | Dispatched: {len(dispatched)} | Skipped: {len(skipped)}")
# Agent status line
agent_line = " | ".join(
f"{'' if v['alive'] else ''}{k}" for k, v in agent_status.items()
)
parts.append(agent_line)
if dispatched:
parts.append("")
parts.append("*Dispatched:*")
for d in dispatched[:5]:
dry = " 🔍" if d.get("dry_run") else ""
parts.append(f" `{d['repo']}#{d['number']}` → {d['agent']}{dry}")
# Top unassigned
unassigned = [i for i in backlog if not i["assignees"]][:3]
if unassigned:
parts.append("")
parts.append("*Top unassigned:*")
for i in unassigned:
parts.append(f" [{i['score']}] `{i['repo']}#{i['number']}` {i['title'][:40]}")
return "\n".join(parts)
# ---------------------------------------------------------------------------
# 6. MAIN
# ---------------------------------------------------------------------------
def run_cycle(dry_run=False):
"""Execute one full orchestration cycle."""
global GITEA_TOKEN, TELEGRAM_TOKEN
GITEA_TOKEN = load_gitea_token()
TELEGRAM_TOKEN = load_telegram_token()
print("\n[1/4] Reading backlog...")
backlog = read_backlog()
print("\n[2/4] Scoring and prioritizing...")
backlog = prioritize_backlog(backlog)
for i in backlog[:10]:
print(f" [{i['score']:3d}] {i['repo']}/{i['number']}: {i['title'][:55]}")
print("\n[3/4] Checking agent health...")
agent_status = get_agent_status()
print("\n[4/4] Dispatching...")
dispatched, skipped = dispatch_cycle(backlog, agent_status, dry_run=dry_run)
# Generate reports
report = generate_report(backlog, dispatched, skipped, agent_status, dry_run=dry_run)
print("\n" + report)
# Send Telegram notification
if dispatched or not dry_run:
tg_msg = format_telegram_report(backlog, dispatched, skipped, agent_status, dry_run=dry_run)
send_telegram(tg_msg)
return backlog, dispatched, skipped
def main():
import argparse
parser = argparse.ArgumentParser(description="Sovereign Orchestrator v1")
parser.add_argument("--once", action="store_true", help="Single dispatch cycle")
parser.add_argument("--daemon", action="store_true", help="Run every 15 min")
parser.add_argument("--dry-run", action="store_true", help="Score/report only, no dispatch")
parser.add_argument("--interval", type=int, default=DAEMON_INTERVAL,
help=f"Daemon interval in seconds (default: {DAEMON_INTERVAL})")
args = parser.parse_args()
if not any([args.once, args.daemon, args.dry_run]):
args.dry_run = True # safe default
print("[INFO] No mode specified, defaulting to --dry-run")
print("=" * 60)
print(" SOVEREIGN ORCHESTRATOR v1")
print("=" * 60)
if args.daemon:
print(f"[DAEMON] Running every {args.interval}s (Ctrl+C to stop)")
cycle = 0
while True:
cycle += 1
print(f"\n--- Cycle {cycle} ---")
try:
run_cycle(dry_run=args.dry_run)
except Exception as e:
print(f"[ERROR] Cycle failed: {e}")
print(f"[DAEMON] Sleeping {args.interval}s...")
time.sleep(args.interval)
else:
run_cycle(dry_run=args.dry_run)
if __name__ == "__main__":
main()

526
scripts/kaizen_retro.py Normal file
View File

@@ -0,0 +1,526 @@
#!/usr/bin/env python3
"""
Kaizen Retro — Automated retrospective after every burn cycle.
Reads overnight Gitea activity, fleet state, and loop logs.
Generates ONE concrete improvement suggestion and posts it.
Usage:
python3 scripts/kaizen_retro.py [--dry-run]
"""
from __future__ import annotations
import argparse
import json
import os
import sys
import urllib.error
import urllib.request
from datetime import datetime, timedelta, timezone
from pathlib import Path
from typing import Any, Optional
# Ensure repo root is on path so we can import gitea_client
REPO_ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(REPO_ROOT))
from gitea_client import GiteaClient, GiteaError
# ---------------------------------------------------------------------------
# Configuration
# ---------------------------------------------------------------------------
REPOS = [
"Timmy_Foundation/the-nexus",
"Timmy_Foundation/timmy-config",
"Timmy_Foundation/timmy-home",
"Timmy_Foundation/the-door",
"Timmy_Foundation/turboquant",
"Timmy_Foundation/hermes-agent",
"Timmy_Foundation/.profile",
]
HERMES_HOME = Path.home() / ".hermes"
TIMMY_HOME = Path.home() / ".timmy"
WORKFORCE_STATE_PATH = HERMES_HOME / "workforce-state.json"
FLEET_ROUTING_PATH = HERMES_HOME / "fleet-routing.json"
CHANNEL_DIR_PATH = REPO_ROOT / "channel_directory.json"
REPORTS_DIR = REPO_ROOT / "reports"
MORNING_REPORT_REPO = "Timmy_Foundation/timmy-config"
TELEGRAM_BOT_TOKEN = os.environ.get("TELEGRAM_BOT_TOKEN")
TELEGRAM_CHAT_ID = os.environ.get("TELEGRAM_HOME_CHANNEL", "-1003664764329")
TELEGRAM_MAX_LEN = 4000 # leave headroom below the 4096 hard limit
STALE_DAYS = 7
MAX_ATTEMPT_COMMENT_THRESHOLD = 5
ISSUE_TYPE_KEYWORDS = {
"bug": ["bug", "fix", "crash", "error", "regression", "broken"],
"feature": ["feature", "implement", "add", "support", "enable"],
"docs": ["doc", "readme", "wiki", "guide", "documentation"],
"kaizen": ["kaizen", "retro", "improvement", "continuous"],
"devops": ["deploy", "ci", "cd", "docker", "server", "infra"],
}
BLOCKER_LABELS = {"blocked", "timeout", "stale", "help wanted", "wontfix", "duplicate"}
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def load_json(path: Path) -> Any:
if not path.exists():
return None
with open(path) as f:
return json.load(f)
def iso_day_ago(days: int = 1) -> str:
return (datetime.now(timezone.utc) - timedelta(days=days)).isoformat()
def classify_issue_type(issue: dict) -> str:
title = (issue.get("title", "") or "").lower()
body = (issue.get("body", "") or "").lower()
labels = [l.get("name", "").lower() for l in issue.get("labels", []) or []]
text = f"{title} {body} {' '.join(labels)}"
words = set(text.split())
best = "other"
best_score = 0
for kind, keywords in ISSUE_TYPE_KEYWORDS.items():
# Short keywords (<=3 chars) require whole-word match to avoid false positives like
# "ci" inside "cleanup" or "cd" inside "abcde".
score = sum(
1 for kw in keywords
if (len(kw) <= 3 and kw in words) or (len(kw) > 3 and kw in text)
)
# label match is stronger
for label in labels:
label_words = set(label.split())
if any(
(len(kw) <= 3 and kw in label_words) or (len(kw) > 3 and kw in label)
for kw in keywords
):
score += 3
if score > best_score:
best_score = score
best = kind
return best
def is_max_attempts_candidate(issue: dict) -> bool:
"""Heuristic for issues that consumed excessive attempts."""
labels = {l.get("name", "").lower() for l in issue.get("labels", []) or []}
if labels & BLOCKER_LABELS:
return True
if issue.get("comments", 0) >= MAX_ATTEMPT_COMMENT_THRESHOLD:
return True
created = issue.get("created_at")
if created:
try:
created_dt = datetime.fromisoformat(created.replace("Z", "+00:00"))
if datetime.now(timezone.utc) - created_dt > timedelta(days=STALE_DAYS):
return True
except Exception:
pass
return False
def telegram_send(text: str, bot_token: str, chat_id: str) -> list[dict]:
"""Post text to Telegram, chunking if it exceeds the message limit."""
url = f"https://api.telegram.org/bot{bot_token}/sendMessage"
chunks = []
if len(text) <= TELEGRAM_MAX_LEN:
chunks = [text]
else:
# Split on newlines to preserve readability
lines = text.splitlines(keepends=True)
current = ""
for line in lines:
if len(current) + len(line) > TELEGRAM_MAX_LEN:
if current:
chunks.append(current)
current = line
else:
current += line
if current:
chunks.append(current)
results = []
for i, chunk in enumerate(chunks):
prefix = f"*(part {i + 1}/{len(chunks)})*\n" if len(chunks) > 1 else ""
payload = {"chat_id": chat_id, "text": prefix + chunk, "parse_mode": "Markdown"}
data = json.dumps(payload).encode()
req = urllib.request.Request(url, data=data, headers={"Content-Type": "application/json"})
with urllib.request.urlopen(req, timeout=30) as resp:
results.append(json.loads(resp.read().decode()))
return results
def find_latest_morning_report_issue(client: GiteaClient) -> Optional[int]:
try:
issues = client.list_issues(MORNING_REPORT_REPO, state="open", sort="created", direction="desc", limit=20)
for issue in issues:
if "good morning report" in issue.title.lower() or "morning report" in issue.title.lower():
return issue.number
# fallback to closed
issues = client.list_issues(MORNING_REPORT_REPO, state="closed", sort="created", direction="desc", limit=20)
for issue in issues:
if "good morning report" in issue.title.lower() or "morning report" in issue.title.lower():
return issue.number
except Exception:
pass
return None
def fmt_pct(num: float, den: float) -> str:
if den == 0:
return "N/A"
return f"{num/den:.0%}"
# ---------------------------------------------------------------------------
# Analysis
# ---------------------------------------------------------------------------
def gather_metrics(client: GiteaClient, since: str) -> dict:
"""Collect overnight metrics from Gitea."""
metrics = {
"closed_issues": [],
"merged_prs": [],
"closed_prs": [],
"open_issues": [],
"max_attempts_issues": [],
"by_agent": {},
"by_repo": {},
"by_type": {},
}
for repo in REPOS:
repo_short = repo.split("/")[1]
metrics["by_repo"][repo_short] = {
"closed": 0,
"merged_prs": 0,
"closed_prs": 0,
"open": 0,
"max_attempts": 0,
"successes": 0,
"failures": 0,
}
# Closed issues since window
try:
closed = client.list_issues(repo, state="closed", since=since, sort="updated", direction="desc", limit=100)
for issue in closed:
issue_dict = {
"number": issue.number,
"title": issue.title,
"repo": repo_short,
"type": classify_issue_type({"title": issue.title, "body": issue.body, "labels": [{"name": lb.name} for lb in issue.labels]}),
"assignee": issue.assignees[0].login if issue.assignees else "unassigned",
}
metrics["closed_issues"].append(issue_dict)
metrics["by_repo"][repo_short]["closed"] += 1
metrics["by_repo"][repo_short]["successes"] += 1
agent = issue_dict["assignee"]
if agent not in metrics["by_agent"]:
metrics["by_agent"][agent] = {"successes": 0, "failures": 0, "closed": 0, "repos": set()}
metrics["by_agent"][agent]["successes"] += 1
metrics["by_agent"][agent]["closed"] += 1
metrics["by_agent"][agent]["repos"].add(repo_short)
t = issue_dict["type"]
if t not in metrics["by_type"]:
metrics["by_type"][t] = {"successes": 0, "failures": 0, "total": 0}
metrics["by_type"][t]["successes"] += 1
metrics["by_type"][t]["total"] += 1
except Exception as exc:
print(f"Warning: could not load closed issues for {repo}: {exc}", file=sys.stderr)
# Open issues (for stale / max-attempts detection)
try:
open_issues = client.list_issues(repo, state="open", sort="created", direction="desc", limit=100)
metrics["by_repo"][repo_short]["open"] = len(open_issues)
for issue in open_issues:
issue_raw = {
"number": issue.number,
"title": issue.title,
"labels": [{"name": lb.name} for lb in issue.labels],
"comments": issue.comments,
"created_at": issue.created_at,
}
if is_max_attempts_candidate(issue_raw):
metrics["max_attempts_issues"].append({
"number": issue.number,
"title": issue.title,
"repo": repo_short,
"type": classify_issue_type({"title": issue.title, "body": issue.body, "labels": issue_raw["labels"]}),
"assignee": issue.assignees[0].login if issue.assignees else "unassigned",
})
metrics["by_repo"][repo_short]["max_attempts"] += 1
metrics["by_repo"][repo_short]["failures"] += 1
agent = issue.assignees[0].login if issue.assignees else "unassigned"
if agent not in metrics["by_agent"]:
metrics["by_agent"][agent] = {"successes": 0, "failures": 0, "closed": 0, "repos": set()}
metrics["by_agent"][agent]["failures"] += 1
metrics["by_agent"][agent]["repos"].add(repo_short)
t = classify_issue_type({"title": issue.title, "body": issue.body, "labels": issue_raw["labels"]})
if t not in metrics["by_type"]:
metrics["by_type"][t] = {"successes": 0, "failures": 0, "total": 0}
metrics["by_type"][t]["failures"] += 1
metrics["by_type"][t]["total"] += 1
except Exception as exc:
print(f"Warning: could not load open issues for {repo}: {exc}", file=sys.stderr)
# PRs merged / closed since window (filter client-side; Gitea PR API ignores since)
try:
prs = client.list_pulls(repo, state="closed", sort="updated", limit=100)
since_dt = datetime.fromisoformat(since.replace("Z", "+00:00"))
for pr in prs:
updated = pr.updated_at or pr.created_at or ""
try:
updated_dt = datetime.fromisoformat(updated.replace("Z", "+00:00"))
if updated_dt < since_dt:
continue
except Exception:
pass
if pr.merged:
metrics["merged_prs"].append({
"number": pr.number,
"title": pr.title,
"repo": repo_short,
"user": pr.user.login if pr.user else "unknown",
})
metrics["by_repo"][repo_short]["merged_prs"] += 1
else:
metrics["closed_prs"].append({
"number": pr.number,
"title": pr.title,
"repo": repo_short,
"user": pr.user.login if pr.user else "unknown",
})
metrics["by_repo"][repo_short]["closed_prs"] += 1
except Exception as exc:
print(f"Warning: could not load PRs for {repo}: {exc}", file=sys.stderr)
# Convert sets to lists for JSON serialization
for agent in metrics["by_agent"].values():
agent["repos"] = sorted(agent["repos"])
return metrics
def load_workforce_state() -> dict:
return load_json(WORKFORCE_STATE_PATH) or {}
def load_fleet_routing() -> list[dict]:
data = load_json(FLEET_ROUTING_PATH)
if data and "agents" in data:
return data["agents"]
return []
def generate_suggestion(metrics: dict, fleet: list[dict]) -> str:
"""Generate ONE concrete improvement suggestion based on the data."""
by_agent = metrics["by_agent"]
by_repo = metrics["by_repo"]
by_type = metrics["by_type"]
max_attempts = metrics["max_attempts_issues"]
suggestions: list[str] = []
# 1. Agent with poor repo performance
for agent, stats in by_agent.items():
total = stats["successes"] + stats["failures"]
if total >= 3 and stats["successes"] == 0:
repos = ", ".join(stats["repos"])
suggestions.append(
f"🎯 **{agent}** has a 0% verify rate over the last cycle (0/{total}) on repos: {repos}. "
f"Consider removing these repos from {agent}'s routing or providing targeted onboarding."
)
# 2. Repo with highest failure concentration
repo_failures = [(r, s) for r, s in by_repo.items() if s["failures"] > 0]
if repo_failures:
repo_failures.sort(key=lambda x: x[1]["failures"], reverse=True)
worst_repo, worst_stats = repo_failures[0]
total_repo = worst_stats["successes"] + worst_stats["failures"]
if worst_stats["failures"] >= 2:
suggestions.append(
f"🎯 **{worst_repo}** has the most friction ({worst_stats['failures']} blocked/stale issues, "
f"{fmt_pct(worst_stats['successes'], total_repo)} success). "
f"Consider splitting issues in {worst_repo} into smaller chunks or assigning a stronger agent."
)
# 3. Max-attempts pattern
if len(max_attempts) >= 3:
type_counts: dict[str, int] = {}
for issue in max_attempts:
type_counts[issue["type"]] = type_counts.get(issue["type"], 0) + 1
top_type = max(type_counts, key=type_counts.get) if type_counts else "unknown"
suggestions.append(
f"🎯 **{len(max_attempts)} issues** hit max-attempts or went stale. "
f"The dominant type is **{top_type}**. "
f"Consider adding acceptance criteria templates or pre-flight checklists for {top_type} issues."
)
# 4. Issue type disparity
for t, stats in by_type.items():
total = stats["total"]
if total >= 3 and stats["successes"] == 0:
suggestions.append(
f"🎯 **{t}** issues have a 0% closure rate ({stats['failures']} stale). "
f"Consider routing all {t} issues to a specialist agent or creating a dedicated playbook."
)
# 5. Fleet routing gap (if fleet data exists)
active_agents = {a["name"] for a in fleet if a.get("active")}
assigned_agents = set(by_agent.keys())
idle_agents = active_agents - assigned_agents - {"unassigned"}
if len(idle_agents) >= 2:
suggestions.append(
f"🎯 **{len(idle_agents)} active agents** have no assignments this cycle: {', '.join(idle_agents)}. "
f"Consider expanding their repo lists or investigating why they aren't receiving work."
)
if suggestions:
return suggestions[0]
# Fallback: celebrate or nudge
total_closed = len(metrics["closed_issues"])
total_merged = len(metrics["merged_prs"])
if total_closed >= 5 or total_merged >= 3:
return (
f"🎯 Strong cycle: {total_closed} issues closed, {total_merged} PRs merged. "
f"Next improvement: write down the top 3 patterns that made this cycle successful so we can replicate them."
)
return (
"🎯 Low activity this cycle. Next improvement: ensure at least one agent loop is actively polling "
"for unassigned issues so work doesn't sit idle."
)
def build_report(metrics: dict, suggestion: str, since: str) -> str:
now = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M UTC")
period = since[:10]
lines = [
f"# 🌀 Kaizen Retro — {now}",
f"*Period: {period} → now*\n",
"## Numbers",
f"- **Issues closed:** {len(metrics['closed_issues'])}",
f"- **PRs merged:** {len(metrics['merged_prs'])}",
f"- **PRs closed without merge:** {len(metrics['closed_prs'])}",
f"- **Max-attempts / stale issues:** {len(metrics['max_attempts_issues'])}",
"",
"## By Agent",
]
for agent, stats in sorted(metrics["by_agent"].items(), key=lambda x: x[1]["successes"] + x[1]["failures"], reverse=True):
total = stats["successes"] + stats["failures"]
rate = fmt_pct(stats["successes"], total)
lines.append(f"- **{agent}**: {stats['successes']} closed, {stats['failures']} stale / max-attempts — verify rate {rate}")
lines.extend(["", "## By Repo"])
for repo, stats in sorted(metrics["by_repo"].items(), key=lambda x: x[1]["successes"] + x[1]["failures"], reverse=True):
total = stats["successes"] + stats["failures"]
if total == 0 and stats["open"] == 0:
continue
rate = fmt_pct(stats["successes"], total)
lines.append(
f"- **{repo}**: {stats['successes']} closed, {stats['failures']} stale, {stats['open']} open — verify rate {rate}"
)
lines.extend(["", "## By Issue Type"])
for t, stats in sorted(metrics["by_type"].items(), key=lambda x: x[1]["total"], reverse=True):
total = stats["total"]
rate = fmt_pct(stats["successes"], total)
lines.append(f"- **{t}**: {stats['successes']} closed, {stats['failures']} stale — verify rate {rate}")
if metrics["max_attempts_issues"]:
lines.extend(["", "## Max-Attempts / Stale Issues"])
for issue in metrics["max_attempts_issues"][:10]:
lines.append(f"- {issue['repo']}#{issue['number']} ({issue['type']}, assignee: {issue['assignee']}) — {issue['title']}")
if len(metrics["max_attempts_issues"]) > 10:
lines.append(f"- … and {len(metrics['max_attempts_issues']) - 10} more")
lines.extend(["", "## One Concrete Improvement", suggestion, ""])
return "\n".join(lines)
# ---------------------------------------------------------------------------
# Main
# ---------------------------------------------------------------------------
def main() -> int:
parser = argparse.ArgumentParser(description="Kaizen Retro — automated burn-cycle retrospective")
parser.add_argument("--dry-run", action="store_true", help="Print report but do not post")
parser.add_argument("--since", type=str, help="ISO timestamp for lookback window (default: 24h ago)")
parser.add_argument("--post-to", type=str, help="Override Telegram chat ID")
args = parser.parse_args()
since = args.since or iso_day_ago(1)
client = GiteaClient()
print("Gathering metrics since", since)
metrics = gather_metrics(client, since)
fleet = load_fleet_routing()
suggestion = generate_suggestion(metrics, fleet)
report = build_report(metrics, suggestion, since)
print(report)
# Save JSON snapshot
REPORTS_DIR.mkdir(parents=True, exist_ok=True)
snapshot_path = REPORTS_DIR / f"kaizen-retro-{datetime.now(timezone.utc).strftime('%Y%m%d')}.json"
snapshot = {
"generated_at": datetime.now(timezone.utc).isoformat(),
"since": since,
"metrics": metrics,
"suggestion": suggestion,
"report_markdown": report,
}
with open(snapshot_path, "w") as f:
json.dump(snapshot, f, indent=2)
print(f"\nSnapshot saved to {snapshot_path}")
if args.dry_run:
return 0
# Post to Telegram
chat_id = args.post_to or TELEGRAM_CHAT_ID
bot_token = TELEGRAM_BOT_TOKEN
if bot_token and chat_id:
try:
telegram_send(report, bot_token, chat_id)
print("Posted to Telegram.")
except Exception as exc:
print(f"Failed to post to Telegram: {exc}", file=sys.stderr)
else:
print("Telegram not configured (set TELEGRAM_BOT_TOKEN and TELEGRAM_HOME_CHANNEL).", file=sys.stderr)
# Comment on latest morning report issue
morning_issue = find_latest_morning_report_issue(client)
if morning_issue:
try:
client.create_comment(MORNING_REPORT_REPO, morning_issue, report)
print(f"Commented on morning report issue #{morning_issue}.")
except Exception as exc:
print(f"Failed to comment on morning report issue: {exc}", file=sys.stderr)
else:
print("No morning report issue found to comment on.", file=sys.stderr)
return 0
if __name__ == "__main__":
sys.exit(main())

112
tasks.py
View File

@@ -1860,22 +1860,56 @@ def good_morning_report():
except Exception:
pass
# Genchi Genbutsu: count verified completions from the last 24h
verified_completions = 0
raw_completions = 0
metrics_dir = Path.home() / ".hermes" / "logs"
for metrics_file in metrics_dir.glob("*-metrics.jsonl"):
try:
with open(metrics_file) as mf:
for line in mf:
line = line.strip()
if not line:
continue
try:
row = json.loads(line)
except json.JSONDecodeError:
continue
ts = row.get("ts", "")
if not ts:
continue
try:
from datetime import datetime as _dt, timezone as _tz, timedelta as _td
row_time = _dt.fromisoformat(ts.replace("Z", "+00:00"))
if (now - row_time) > _td(hours=24):
continue
except Exception:
continue
if row.get("outcome") != "success":
continue
raw_completions += 1
if row.get("verified") is True:
verified_completions += 1
except Exception:
pass
# --- BUILD THE REPORT ---
body = f"""Good morning, Alexander. It's {day_name}.
## Overnight Debrief
**Heartbeat:** {tick_count} ticks logged overnight.
**Gitea:** {"up all night" if gitea_up else "⚠️ had downtime"}
**Local inference:** {"running steady" if local_inference_up else "⚠️ had downtime"}
**Model status:** {model_status}
**Models on disk:** {len(models_loaded)} ({', '.join(m for m in models_loaded if 'timmy' in m.lower() or 'hermes' in m.lower()) or 'none with our name'})
**Alerts:** {len(alerts)} {'' + '; '.join(alerts[-3:]) if alerts else '(clean night)'}
|**Heartbeat:** {tick_count} ticks logged overnight.
|**Gitea:** {"up all night" if gitea_up else "⚠️ had downtime"}
|**Local inference:** {"running steady" if local_inference_up else "⚠️ had downtime"}
|**Model status:** {model_status}
|**Models on disk:** {len(models_loaded)} ({', '.join(m for m in models_loaded if 'timmy' in m.lower() or 'hermes' in m.lower()) or 'none with our name'})
|**Alerts:** {len(alerts)} {'' + '; '.join(alerts[-3:]) if alerts else '(clean night)'}
{briefing_summary}
**DPO training pairs staged:** {dpo_count} session files exported
**Local model smoke test:** {smoke_result}
**Verified completions (24h):** {verified_completions} {'(Genchi Genbutsu clean)' if verified_completions == raw_completions else f'({raw_completions - verified_completions} raw completions failed verification)'}
## Gitea Pulse
@@ -1915,6 +1949,29 @@ That's all. Have a good morning.
return {"filed": False, "error": str(e)}
# ── NEW 6b: Kaizen Retro ─────────────────────────────────────────────
@huey.periodic_task(crontab(hour="7", minute="15")) # 7:15 AM daily, after morning report
def kaizen_retro():
"""Run the automated burn-cycle retrospective."""
retro_script = Path(__file__).resolve().parent / "bin" / "kaizen-retro.sh"
if not retro_script.exists():
return {"ran": False, "error": "kaizen-retro.sh not found"}
result = subprocess.run(
["bash", str(retro_script)],
capture_output=True,
text=True,
timeout=300,
)
return {
"ran": True,
"exit_code": result.returncode,
"stdout": result.stdout[-2000:] if result.stdout else "",
"stderr": result.stderr[-1000:] if result.stderr else "",
}
# ── NEW 7: Repo Watchdog ─────────────────────────────────────────────
@huey.periodic_task(crontab(minute="*/20")) # every 20 minutes
@@ -2323,7 +2380,38 @@ def velocity_tracking():
total_open += open_n
total_closed += closed_n
results.append({"repo": repo, "open": open_n, "closed": closed_n, "date": today})
data = {"date": today, "repos": results, "total_open": total_open, "total_closed": total_closed}
# Genchi Genbutsu: count verified completions from agent metrics
verified_completions = 0
raw_completions = 0
metrics_dir = Path.home() / ".hermes" / "logs"
for metrics_file in metrics_dir.glob("*-metrics.jsonl"):
try:
with open(metrics_file) as mf:
for line in mf:
line = line.strip()
if not line:
continue
try:
row = json.loads(line)
except json.JSONDecodeError:
continue
if row.get("outcome") != "success":
continue
raw_completions += 1
if row.get("verified") is True:
verified_completions += 1
except Exception:
pass
data = {
"date": today,
"repos": results,
"total_open": total_open,
"total_closed": total_closed,
"raw_completions": raw_completions,
"verified_completions": verified_completions,
}
with open(report_file, "w") as f:
json.dump(data, f, indent=2)
# Dashboard
@@ -2333,14 +2421,16 @@ def velocity_tracking():
for r in results:
f.write(f"| {r['repo'].split('/')[-1]} | {r['open']} | {r['closed']} |\n")
f.write(f"| **TOTAL** | **{total_open}** | **{total_closed}** |\n\n")
f.write(f"**Verified completions (Genchi Genbutsu):** {verified_completions}\n")
f.write(f"**Raw completions:** {raw_completions}\n\n")
# Trend
prior = sorted(glob.glob(os.path.join(report_dir, "velocity-*.json")))
if len(prior) > 1:
f.write("## Recent Trend\n\n| Date | Total Open | Total Closed |\n|---|---|---|\n")
f.write("## Recent Trend\n\n| Date | Total Open | Total Closed | Verified |\n|---|---|---|---|\n")
for pf in prior[-10:]:
pd = json.load(open(pf))
f.write(f"| {pd['date']} | {pd['total_open']} | {pd['total_closed']} |\n")
msg = f"Velocity: {total_open} open, {total_closed} closed ({today})"
f.write(f"| {pd['date']} | {pd['total_open']} | {pd['total_closed']} | {pd.get('verified_completions', '-')} |\n")
msg = f"Velocity: {total_open} open, {total_closed} closed, {verified_completions} verified ({today})"
if len(prior) > 1:
prev = json.load(open(prior[-2]))
if total_open > prev["total_open"]:

283
tests/test_kaizen_retro.py Normal file
View File

@@ -0,0 +1,283 @@
"""Tests for the Kaizen Retro burn-cycle retrospective script."""
from __future__ import annotations
import importlib.util
import json
import sys
from datetime import datetime, timedelta, timezone
from pathlib import Path
from unittest.mock import MagicMock, patch
import pytest
REPO_ROOT = Path(__file__).parent.parent
# Load kaizen_retro.py as a module (it lives in scripts/, not a package)
spec = importlib.util.spec_from_file_location("kaizen_retro", REPO_ROOT / "scripts" / "kaizen_retro.py")
kr = importlib.util.module_from_spec(spec)
spec.loader.exec_module(kr)
# ── classify_issue_type ───────────────────────────────────────────────────
class TestClassifyIssueType:
def test_classifies_bug_from_title(self):
issue = {"title": "Fix crash on startup", "body": "", "labels": []}
assert kr.classify_issue_type(issue) == "bug"
def test_classifies_feature_from_label(self):
issue = {"title": "Add dark mode", "body": "", "labels": [{"name": "enhancement"}]}
# label "enhancement" doesn't match any keyword directly, but "feature" and "add" are in title
assert kr.classify_issue_type(issue) == "feature"
def test_classifies_docs_from_label(self):
issue = {"title": "Update guide", "body": "", "labels": [{"name": "documentation"}]}
assert kr.classify_issue_type(issue) == "docs"
def test_label_match_stronger_than_title(self):
issue = {"title": "Something random", "body": "", "labels": [{"name": "bug"}]}
assert kr.classify_issue_type(issue) == "bug"
def test_kaizen_takes_precedence_with_both_labels(self):
issue = {"title": "Process improvement", "body": "", "labels": [{"name": "kaizen"}, {"name": "bug"}]}
# kaizen label gives +3, bug gives +3, tie goes to first seen? kaizen appears first in dict
assert kr.classify_issue_type(issue) == "kaizen"
def test_defaults_to_other(self):
issue = {"title": "Tidy up naming", "body": "No user-facing change", "labels": [{"name": "cleanup"}]}
assert kr.classify_issue_type(issue) == "other"
# ── is_max_attempts_candidate ─────────────────────────────────────────────
class TestIsMaxAttemptsCandidate:
def test_blocker_label_returns_true(self):
issue = {"labels": [{"name": "blocked"}], "comments": 0, "created_at": "2026-04-07T00:00:00Z"}
assert kr.is_max_attempts_candidate(issue) is True
def test_timeout_label_returns_true(self):
issue = {"labels": [{"name": "timeout"}], "comments": 0, "created_at": "2026-04-07T00:00:00Z"}
assert kr.is_max_attempts_candidate(issue) is True
def test_high_comment_count_returns_true(self):
issue = {"labels": [], "comments": 5, "created_at": "2026-04-07T00:00:00Z"}
assert kr.is_max_attempts_candidate(issue) is True
def test_fresh_issue_with_low_comments_returns_false(self):
now = datetime.now(timezone.utc)
issue = {"labels": [], "comments": 2, "created_at": now.isoformat()}
assert kr.is_max_attempts_candidate(issue) is False
def test_stale_age_returns_true(self):
old = datetime.now(timezone.utc) - timedelta(days=10)
issue = {"labels": [], "comments": 0, "created_at": old.isoformat()}
assert kr.is_max_attempts_candidate(issue) is True
# ── fmt_pct ───────────────────────────────────────────────────────────────
class TestFmtPct:
def test_basic_percentage(self):
assert kr.fmt_pct(3, 4) == "75%"
def test_zero_denominator(self):
assert kr.fmt_pct(0, 0) == "N/A"
def test_perfect_rate(self):
assert kr.fmt_pct(10, 10) == "100%"
# ── generate_suggestion ───────────────────────────────────────────────────
class TestGenerateSuggestion:
def test_agent_zero_success_rate(self):
metrics = {
"by_agent": {
"groq": {"successes": 0, "failures": 5, "closed": 0, "repos": ["timmy-home"]},
},
"by_repo": {},
"by_type": {},
"max_attempts_issues": [],
"closed_issues": [],
"merged_prs": [],
"closed_prs": [],
}
suggestion = kr.generate_suggestion(metrics, [])
assert "groq" in suggestion
assert "0%" in suggestion or "verify rate" in suggestion
def test_repo_with_most_failures(self):
metrics = {
"by_agent": {},
"by_repo": {
"the-nexus": {"successes": 2, "failures": 5, "closed": 2, "open": 3},
},
"by_type": {},
"max_attempts_issues": [],
"closed_issues": [],
"merged_prs": [],
"closed_prs": [],
}
suggestion = kr.generate_suggestion(metrics, [])
assert "the-nexus" in suggestion
assert "friction" in suggestion
def test_max_attempts_pattern(self):
metrics = {
"by_agent": {},
"by_repo": {},
"by_type": {},
"max_attempts_issues": [
{"type": "devops"}, {"type": "devops"}, {"type": "feature"}
],
"closed_issues": [],
"merged_prs": [],
"closed_prs": [],
}
suggestion = kr.generate_suggestion(metrics, [])
assert "devops" in suggestion
assert "max-attempts" in suggestion.lower() or "stale" in suggestion.lower()
def test_idle_agents(self):
metrics = {
"by_agent": {},
"by_repo": {},
"by_type": {},
"max_attempts_issues": [],
"closed_issues": [],
"merged_prs": [],
"closed_prs": [],
}
fleet = [{"name": "allegro", "active": True}, {"name": "ezra", "active": True}]
suggestion = kr.generate_suggestion(metrics, fleet)
assert "idle" in suggestion.lower() or "no assignments" in suggestion.lower()
def test_fallback_celebration(self):
metrics = {
"by_agent": {},
"by_repo": {},
"by_type": {},
"max_attempts_issues": [],
"closed_issues": [{}, {}, {}, {}, {}],
"merged_prs": [{}, {}, {}],
"closed_prs": [],
}
suggestion = kr.generate_suggestion(metrics, [])
assert "Strong cycle" in suggestion
def test_fallback_low_activity(self):
metrics = {
"by_agent": {},
"by_repo": {},
"by_type": {},
"max_attempts_issues": [],
"closed_issues": [],
"merged_prs": [],
"closed_prs": [],
}
suggestion = kr.generate_suggestion(metrics, [])
assert "Low activity" in suggestion or "idle" in suggestion.lower()
# ── build_report ──────────────────────────────────────────────────────────
class TestBuildReport:
def test_report_contains_numbers_section(self):
metrics = {
"closed_issues": [{}, {}],
"merged_prs": [{}],
"closed_prs": [],
"max_attempts_issues": [],
"by_agent": {"ezra": {"successes": 2, "failures": 0, "repos": ["timmy-config"]}},
"by_repo": {"timmy-config": {"successes": 2, "failures": 0, "open": 1}},
"by_type": {"feature": {"successes": 2, "failures": 0, "total": 2}},
}
report = kr.build_report(metrics, "Do better.", "2026-04-06T00:00:00+00:00")
assert "## Numbers" in report
assert "Issues closed:** 2" in report
assert "PRs merged:** 1" in report
assert "## By Agent" in report
assert "## By Repo" in report
assert "## By Issue Type" in report
assert "Do better." in report
def test_report_skips_empty_repos(self):
metrics = {
"closed_issues": [],
"merged_prs": [],
"closed_prs": [],
"max_attempts_issues": [],
"by_agent": {},
"by_repo": {"unused-repo": {"successes": 0, "failures": 0, "open": 0}},
"by_type": {},
}
report = kr.build_report(metrics, "Nudge.", "2026-04-06T00:00:00+00:00")
assert "unused-repo" not in report
def test_report_truncates_max_attempts(self):
metrics = {
"closed_issues": [],
"merged_prs": [],
"closed_prs": [],
"max_attempts_issues": [{"repo": "r", "number": i, "type": "bug", "assignee": "a", "title": f"T{i}"} for i in range(15)],
"by_agent": {},
"by_repo": {},
"by_type": {},
}
report = kr.build_report(metrics, "Fix it.", "2026-04-06T00:00:00+00:00")
assert "and 5 more" in report
# ── telegram_send ─────────────────────────────────────────────────────────
class TestTelegramSend:
def test_short_message_sent_in_one_piece(self):
with patch("urllib.request.urlopen") as mock_urlopen:
mock_resp = MagicMock()
mock_resp.read.return_value = b'{"ok": true}'
mock_urlopen.return_value.__enter__.return_value = mock_resp
results = kr.telegram_send("Hello", "fake-token", "123")
assert len(results) == 1
assert results[0]["ok"] is True
# Verify payload
call_args = mock_urlopen.call_args
req = call_args[0][0]
payload = json.loads(req.data.decode())
assert payload["text"] == "Hello"
assert payload["chat_id"] == "123"
def test_long_message_chunked(self):
big_text = "Line\n" * 2000 # ~10k chars
with patch("urllib.request.urlopen") as mock_urlopen:
mock_resp = MagicMock()
mock_resp.read.return_value = b'{"ok": true}'
mock_urlopen.return_value.__enter__.return_value = mock_resp
results = kr.telegram_send(big_text, "fake-token", "123")
assert len(results) >= 2
# First chunk should have a part prefix
req = mock_urlopen.call_args_list[0][0][0]
payload = json.loads(req.data.decode())
assert "(part 1" in payload["text"]
# ── load helpers ──────────────────────────────────────────────────────────
class TestLoadHelpers:
def test_load_json_missing_returns_none(self, tmp_path):
missing = tmp_path / "does_not_exist.json"
assert kr.load_json(missing) is None
def test_load_json_valid(self, tmp_path):
p = tmp_path / "data.json"
p.write_text('{"a": 1}')
assert kr.load_json(p) == {"a": 1}
def test_iso_day_ago_format(self):
s = kr.iso_day_ago(1)
# Should be a valid ISO timestamp string
dt = datetime.fromisoformat(s)
now = datetime.now(timezone.utc)
assert now - dt < timedelta(days=2)