fix(ci): use script file instead of inline Python in playbook-schema

Inline Python code with f-strings and set literals caused YAML parsing
errors. Replaced with reference to scripts/validate_playbook_schema.py.

Refs: #461

.gitea/PULL_REQUEST_TEMPLATE.md

@@ -0,0 +1,54 @@
+## Summary
+
+<!-- What changed and why. One paragraph max. -->
+
+## Governing Issue
+
+<!-- REQUIRED. Every PR must reference at least one issue. Max 3 issues per PR. -->
+<!-- Closes #ISSUENUM -->
+<!-- Refs #ISSUENUM -->
+
+## Acceptance Criteria
+
+<!-- List the specific outcomes this PR delivers. Check each only when proven. -->
+<!-- Copy these from the governing issue if it has them. -->
+
+- [ ] Criterion 1
+- [ ] Criterion 2
+
+## Proof
+
+<!-- No proof = no merge. See CONTRIBUTING.md for the full standard. -->
+
+### Commands / logs / world-state proof
+
+<!-- Paste the exact commands, output, log paths, or world-state artifacts that prove each acceptance criterion was met. -->
+
+```
+$ <command you ran>
+<relevant output>
+```
+
+### Visual proof (if applicable)
+
+<!-- For skin updates, UI changes, dashboard changes: attach screenshot to the PR discussion. -->
+<!-- Name what the screenshot proves. Do not commit binary media unless explicitly required. -->
+
+## Risk and Rollback
+
+<!-- What could go wrong? How do we undo it? -->
+
+- **Risk level:** low / medium / high
+- **What breaks if this is wrong:**
+- **How to rollback:**
+
+## Checklist
+
+<!-- Complete every item before requesting review. -->
+
+- [ ] PR body references at least one issue number (`Closes #N` or `Refs #N`)
+- [ ] Changed files are syntactically valid (`python -c "import ast; ast.parse(open(f).read())"`, `node --check`, `bash -n`)
+- [ ] Proof meets CONTRIBUTING.md standard (exact commands, output, or artifacts — not "looks right")
+- [ ] Branch is up-to-date with base
+- [ ] No more than 3 unrelated issues bundled in this PR
+- [ ] Shell scripts are executable (`chmod +x`)

.gitea/workflows/architecture-lint.yml

@@ -0,0 +1,41 @@
+# architecture-lint.yml — CI gate for the Architecture Linter v2
+# Refs: #437 — repo-aware, test-backed, CI-enforced.
+#
+# Runs on every PR to main.  Validates Python syntax, then runs
+# linter tests and finally lints the repo itself.
+
+name: Architecture Lint
+
+on:
+  pull_request:
+    branches: [main, master]
+  push:
+    branches: [main]
+
+jobs:
+  linter-tests:
+    name: Linter Tests
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - uses: actions/setup-python@v5
+        with:
+          python-version: "3.11"
+      - name: Install test deps
+        run: pip install pytest
+      - name: Compile-check linter
+        run: python3 -m py_compile scripts/architecture_linter_v2.py
+      - name: Run linter tests
+        run: python3 -m pytest tests/test_linter.py -v
+
+  lint-repo:
+    name: Lint Repository
+    runs-on: ubuntu-latest
+    needs: linter-tests
+    steps:
+      - uses: actions/checkout@v4
+      - uses: actions/setup-python@v5
+        with:
+          python-version: "3.11"
+      - name: Run architecture linter
+        run: python3 scripts/architecture_linter_v2.py .

.gitea/workflows/pr-checklist.yml

@@ -0,0 +1,29 @@
+# pr-checklist.yml — Automated PR quality gate
+# Refs: #393 (PERPLEXITY-08), Epic #385
+#
+# Enforces the review checklist that agents skip when left to self-approve.
+# Runs on every pull_request. Fails fast so bad PRs never reach a reviewer.
+
+name: PR Checklist
+
+on:
+  pull_request:
+    branches: [main, master]
+
+jobs:
+  pr-checklist:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+        with:
+          fetch-depth: 0
+
+      - name: Set up Python
+        uses: actions/setup-python@v5
+        with:
+          python-version: "3.11"
+
+      - name: Run PR checklist
+        env:
+          GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+        run: python3 bin/pr-checklist.py

.gitea/workflows/validate-config.yaml

@@ -0,0 +1,121 @@
+# validate-config.yaml
+# Validates all config files, scripts, and playbooks on every PR.
+# Addresses #289: repo-native validation for timmy-config changes.
+#
+# Runs: YAML lint, Python syntax check, shell lint, JSON validation,
+#       deploy script dry-run, and cron syntax verification.
+
+name: Validate Config
+
+on:
+  pull_request:
+    branches: [main]
+  push:
+    branches: [main]
+
+jobs:
+  yaml-lint:
+    name: YAML Lint
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - name: Install yamllint
+        run: pip install yamllint
+      - name: Lint YAML files
+        run: |
+          find . -name '*.yaml' -o -name '*.yml' | \
+            grep -v '.gitea/workflows' | \
+            xargs -r yamllint -d '{extends: relaxed, rules: {line-length: {max: 200}}}'
+
+  json-validate:
+    name: JSON Validate
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - name: Validate JSON files
+        run: |
+          find . -name '*.json' -print0 | while IFS= read -r -d '' f; do
+            echo "Validating: $f"
+            python3 -m json.tool "$f" > /dev/null || exit 1
+          done
+
+  python-check:
+    name: Python Syntax & Import Check
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - uses: actions/setup-python@v5
+        with:
+          python-version: '3.11'
+      - name: Install dependencies
+        run: |
+          pip install py_compile flake8
+      - name: Compile-check all Python files
+        run: |
+          find . -name '*.py' -print0 | while IFS= read -r -d '' f; do
+            echo "Checking: $f"
+            python3 -m py_compile "$f" || exit 1
+          done
+      - name: Flake8 critical errors only
+        run: |
+          flake8 --select=E9,F63,F7,F82 --show-source --statistics \
+            scripts/ allegro/ cron/ || true
+
+  shell-lint:
+    name: Shell Script Lint
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - name: Install shellcheck
+        run: sudo apt-get install -y shellcheck
+      - name: Lint shell scripts
+        run: |
+          find . -name '*.sh' -print0 | xargs -0 -r shellcheck --severity=error || true
+
+  cron-validate:
+    name: Cron Syntax Check
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - name: Validate cron entries
+        run: |
+          if [ -d cron ]; then
+            find cron -name '*.cron' -o -name '*.crontab' | while read f; do
+              echo "Checking cron: $f"
+              # Basic syntax validation
+              while IFS= read -r line; do
+                [[ "$line" =~ ^#.*$ ]] && continue
+                [[ -z "$line" ]] && continue
+                fields=$(echo "$line" | awk '{print NF}')
+                if [ "$fields" -lt 6 ]; then
+                  echo "ERROR: Too few fields in $f: $line"
+                  exit 1
+                fi
+              done < "$f"
+            done
+          fi
+
+  deploy-dry-run:
+    name: Deploy Script Dry Run
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - name: Syntax-check deploy.sh
+        run: |
+          if [ -f deploy.sh ]; then
+            bash -n deploy.sh
+            echo "deploy.sh syntax OK"
+          fi
+
+  playbook-schema:
+    name: Playbook Schema Validation
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - uses: actions/setup-python@v5
+        with:
+          python-version: '3.11'
+      - name: Install PyYAML
+        run: pip install pyyaml
+      - name: Validate playbook structure
+        run: python3 scripts/validate_playbook_schema.py

.gitignore

@@ -1,9 +1,8 @@
-# Secrets
-*.token
-*.key
-*.secret
-
-# Local state
+*.pyc
+*.pyo
+*.egg-info/
+dist/
+build/
 *.db
 *.db-wal
 *.db-shm
@@ -11,3 +10,27 @@ __pycache__/
 
 # Generated audit reports
 reports/
+
+# Secrets and credentials
+.bash_history
+.git-credentials
+.gitea_token
+.ssh/id_*
+.ssh/known_hosts
+.viminfo
+.wget-hsts
+.profile
+.bashrc
+.bash_logout
+.python_history
+.lesshst
+.selected_editor
+.sudo_as_admin_successful
+.config/telegram/
+.hermes/.env
+.hermes/auth.json
+*.pem
+*.key
+.env
+.env.*
+!.env.example

SOUL.md

@@ -1,3 +1,13 @@
+<!-- 
+  NOTE: This is the BITCOIN INSCRIPTION version of SOUL.md.
+  It is the immutable on-chain conscience. Do not modify this content.
+  
+  The NARRATIVE identity document (for onboarding, Audio Overviews, 
+  and system prompts) lives in timmy-home/SOUL.md.
+  
+  See: #388, #378 for the divergence audit.
+-->
+
 # SOUL.md
 
 ## Inscription 1 — The Immutable Conscience

ansible/BANNED_PROVIDERS.yml

@@ -0,0 +1,47 @@
+# =============================================================================
+# BANNED PROVIDERS — The Timmy Foundation
+# =============================================================================
+# "Anthropic is not only fired, but banned. I don't want these errors
+# cropping up." — Alexander, 2026-04-09
+#
+# This is a HARD BAN. Not deprecated. Not fallback. BANNED.
+# Enforcement: pre-commit hook, linter, Ansible validation, CI tests.
+# =============================================================================
+
+banned_providers:
+  - name: anthropic
+    reason: "Permanently banned. SDK access gated despite active quota. Fleet was bricked because golden state pointed to Anthropic Sonnet."
+    banned_date: "2026-04-09"
+    enforcement: strict  # Ansible playbook FAILS if detected
+    models:
+      - "claude-sonnet-*"
+      - "claude-opus-*"
+      - "claude-haiku-*"
+      - "claude-*"
+    endpoints:
+      - "api.anthropic.com"
+      - "anthropic/*"  # OpenRouter pattern
+    api_keys:
+      - "ANTHROPIC_API_KEY"
+      - "CLAUDE_API_KEY"
+
+# Golden state alternative:
+approved_providers:
+  - name: kimi-coding
+    model: kimi-k2.5
+    role: primary
+  - name: openrouter
+    model: google/gemini-2.5-pro
+    role: fallback
+  - name: ollama
+    model: "gemma4:latest"
+    role: terminal_fallback
+
+# Future evaluation:
+evaluation_candidates:
+  - name: mimo-v2-pro
+    status: pending
+    notes: "Free via Nous Portal for ~2 weeks from 2026-04-07. Add after fallback chain is fixed."
+  - name: hermes-4
+    status: available
+    notes: "Free on Nous Portal. 36B and 70B variants. Home team model."

ansible/README.md

@@ -0,0 +1,95 @@
+# Ansible IaC — The Timmy Foundation Fleet
+
+> One canonical Ansible playbook defines: deadman switch, cron schedule,
+> golden state rollback, agent startup sequence.
+> — KT Final Session 2026-04-08, Priority TWO
+
+## Purpose
+
+This directory contains the **single source of truth** for fleet infrastructure.
+No more ad-hoc recovery implementations. No more overlapping deadman switches.
+No more agents mutating their own configs into oblivion.
+
+**Everything** goes through Ansible. If it's not in a playbook, it doesn't exist.
+
+## Architecture
+
+```
+┌─────────────────────────────────────────────────┐
+│                  Gitea (Source of Truth)          │
+│  timmy-config/ansible/                           │
+│    ├── inventory/hosts.yml    (fleet machines)    │
+│    ├── playbooks/site.yml     (master playbook)   │
+│    ├── roles/                 (reusable roles)    │
+│    └── group_vars/wizards.yml (golden state)      │
+└──────────────────┬──────────────────────────────┘
+                   │  PR merge triggers webhook
+                   ▼
+┌─────────────────────────────────────────────────┐
+│              Gitea Webhook Handler                │
+│  scripts/deploy_on_webhook.sh                     │
+│  → ansible-pull on each target machine            │
+└──────────────────┬──────────────────────────────┘
+                   │  ansible-pull
+                   ▼
+┌──────────┐  ┌──────────┐  ┌──────────┐  ┌──────────┐
+│  Timmy   │  │ Allegro  │  │ Bezalel  │  │  Ezra    │
+│  (Mac)   │  │  (VPS)   │  │  (VPS)   │  │  (VPS)   │
+│          │  │          │  │          │  │          │
+│ deadman  │  │ deadman  │  │ deadman  │  │ deadman  │
+│ cron     │  │ cron     │  │ cron     │  │ cron     │
+│ golden   │  │ golden   │  │ golden   │  │ golden   │
+│ req_log  │  │ req_log  │  │ req_log  │  │ req_log  │
+└──────────┘  └──────────┘  └──────────┘  └──────────┘
+```
+
+## Quick Start
+
+```bash
+# Deploy everything to all machines
+ansible-playbook -i inventory/hosts.yml playbooks/site.yml
+
+# Deploy only golden state config
+ansible-playbook -i inventory/hosts.yml playbooks/golden_state.yml
+
+# Deploy only to a specific wizard
+ansible-playbook -i inventory/hosts.yml playbooks/site.yml --limit bezalel
+
+# Dry run (check mode)
+ansible-playbook -i inventory/hosts.yml playbooks/site.yml --check --diff
+```
+
+## Golden State Provider Chain
+
+All wizard configs converge on this provider chain. **Anthropic is BANNED.**
+
+| Priority | Provider             | Model            | Endpoint                          |
+| -------- | -------------------- | ---------------- | --------------------------------- |
+| 1        | Kimi                 | kimi-k2.5        | https://api.kimi.com/coding/v1    |
+| 2        | Gemini (OpenRouter)  | gemini-2.5-pro   | https://openrouter.ai/api/v1      |
+| 3        | Ollama (local)       | gemma4:latest    | http://localhost:11434/v1         |
+
+## Roles
+
+| Role             | Purpose                                                      |
+| ---------------- | ------------------------------------------------------------ |
+| `wizard_base`    | Common wizard setup: directories, thin config, git pull      |
+| `deadman_switch` | Health check → snapshot good config → rollback on death      |
+| `golden_state`   | Deploy and enforce golden state provider chain               |
+| `request_log`    | SQLite telemetry table for every inference call               |
+| `cron_manager`   | Source-controlled cron jobs — no manual crontab edits         |
+
+## Rules
+
+1. **No manual changes.** If it's not in a playbook, it will be overwritten.
+2. **No Anthropic.** Banned. Enforcement is automated. See `BANNED_PROVIDERS.yml`.
+3. **Idempotent.** Every playbook can run 100 times with the same result.
+4. **PR required.** Config changes go through Gitea PR review, then deploy.
+5. **One identity per machine.** No duplicate agents. Fleet audit enforces this.
+
+## Related Issues
+
+- timmy-config #442: [P2] Ansible IaC Canonical Playbook
+- timmy-config #444: Wire Deadman Switch ACTION
+- timmy-config #443: Thin Config Pattern
+- timmy-config #446: request_log Telemetry Table

ansible/ansible.cfg

@@ -0,0 +1,21 @@
+[defaults]
+inventory = inventory/hosts.yml
+roles_path = roles
+host_key_checking = False
+retry_files_enabled = False
+stdout_callback = yaml
+forks = 10
+timeout = 30
+
+# Logging
+log_path = /var/log/ansible/timmy-fleet.log
+
+[privilege_escalation]
+become = True
+become_method = sudo
+become_user = root
+become_ask_pass = False
+
+[ssh_connection]
+pipelining = True
+ssh_args = -o ControlMaster=auto -o ControlPersist=60s -o StrictHostKeyChecking=no

ansible/inventory/group_vars/wizards.yml

@@ -0,0 +1,74 @@
+# =============================================================================
+# Wizard Group Variables — Golden State Configuration
+# =============================================================================
+# These variables are applied to ALL wizards in the fleet.
+# This IS the golden state. If a wizard deviates, Ansible corrects it.
+# =============================================================================
+
+# --- Deadman Switch ---
+deadman_enabled: true
+deadman_check_interval: 300    # 5 minutes between health checks
+deadman_snapshot_dir: "~/.local/timmy/snapshots"
+deadman_max_snapshots: 10      # Rolling window of good configs
+deadman_restart_cooldown: 60   # Seconds to wait before restart after failure
+deadman_max_restart_attempts: 3
+deadman_escalation_channel: telegram  # Alert Alexander after max attempts
+
+# --- Thin Config ---
+thin_config_path: "~/.timmy/thin_config.yml"
+thin_config_mode: "0444"       # Read-only — agents CANNOT modify
+upstream_repo: "https://forge.alexanderwhitestone.com/Timmy_Foundation/timmy-config.git"
+upstream_branch: main
+config_pull_on_wake: true
+config_validation_enabled: true
+
+# --- Agent Settings ---
+agent_max_turns: 30
+agent_reasoning_effort: high
+agent_verbose: false
+agent_approval_mode: auto
+
+# --- Hermes Harness ---
+hermes_config_dir: "{{ hermes_home }}"
+hermes_bin_dir: "{{ hermes_home }}/bin"
+hermes_skins_dir: "{{ hermes_home }}/skins"
+hermes_playbooks_dir: "{{ hermes_home }}/playbooks"
+hermes_memories_dir: "{{ hermes_home }}/memories"
+
+# --- Request Log (Telemetry) ---
+request_log_enabled: true
+request_log_path: "~/.local/timmy/request_log.db"
+request_log_rotation_days: 30  # Archive logs older than 30 days
+request_log_sync_to_gitea: false  # Future: push telemetry summaries to Gitea
+
+# --- Cron Schedule ---
+# All cron jobs are managed here. No manual crontab edits.
+cron_jobs:
+  - name: "Deadman health check"
+    job: "cd {{ wizard_home }}/workspace/timmy-config && python3 fleet/health_check.py"
+    minute: "*/5"
+    hour: "*"
+    enabled: "{{ deadman_enabled }}"
+
+  - name: "Muda audit"
+    job: "cd {{ wizard_home }}/workspace/timmy-config && bash fleet/muda-audit.sh >> /tmp/muda-audit.log 2>&1"
+    minute: "0"
+    hour: "21"
+    weekday: "0"
+    enabled: true
+
+  - name: "Config pull from upstream"
+    job: "cd {{ wizard_home }}/workspace/timmy-config && git pull --ff-only origin main"
+    minute: "*/15"
+    hour: "*"
+    enabled: "{{ config_pull_on_wake }}"
+
+  - name: "Request log rotation"
+    job: "python3 -c \"import sqlite3,datetime; db=sqlite3.connect('{{ request_log_path }}'); db.execute('DELETE FROM request_log WHERE timestamp < datetime(\\\"now\\\", \\\"-{{ request_log_rotation_days }} days\\\")'); db.commit()\""
+    minute: "0"
+    hour: "3"
+    enabled: "{{ request_log_enabled }}"
+
+# --- Provider Enforcement ---
+# These are validated on every Ansible run. Any Anthropic reference = failure.
+provider_ban_enforcement: strict  # strict = fail playbook, warn = log only

ansible/inventory/hosts.yml

@@ -0,0 +1,119 @@
+# =============================================================================
+# Fleet Inventory — The Timmy Foundation
+# =============================================================================
+# Source of truth for all machines in the fleet.
+# Update this file when machines are added/removed.
+# All changes go through PR review.
+# =============================================================================
+
+all:
+  children:
+    wizards:
+      hosts:
+        timmy:
+          ansible_host: localhost
+          ansible_connection: local
+          wizard_name: Timmy
+          wizard_role: "Primary wizard — soul of the fleet"
+          wizard_provider_primary: kimi-coding
+          wizard_model_primary: kimi-k2.5
+          hermes_port: 8081
+          api_port: 8645
+          wizard_home: "{{ ansible_env.HOME }}/wizards/timmy"
+          hermes_home: "{{ ansible_env.HOME }}/.hermes"
+          machine_type: mac
+          # Timmy runs on Alexander's M3 Max
+          ollama_available: true
+
+        allegro:
+          ansible_host: 167.99.126.228
+          ansible_user: root
+          wizard_name: Allegro
+          wizard_role: "Kimi-backed third wizard house — tight coding tasks"
+          wizard_provider_primary: kimi-coding
+          wizard_model_primary: kimi-k2.5
+          hermes_port: 8081
+          api_port: 8645
+          wizard_home: /root/wizards/allegro
+          hermes_home: /root/.hermes
+          machine_type: vps
+          ollama_available: false
+
+        bezalel:
+          ansible_host: 159.203.146.185
+          ansible_user: root
+          wizard_name: Bezalel
+          wizard_role: "Forge-and-testbed wizard — infrastructure, deployment, hardening"
+          wizard_provider_primary: kimi-coding
+          wizard_model_primary: kimi-k2.5
+          hermes_port: 8081
+          api_port: 8656
+          wizard_home: /root/wizards/bezalel
+          hermes_home: /root/.hermes
+          machine_type: vps
+          ollama_available: false
+          # NOTE: The awake Bezalel may be the duplicate.
+          # Fleet audit (the-nexus #1144) will resolve identity.
+
+        ezra:
+          ansible_host: 143.198.27.163
+          ansible_user: root
+          wizard_name: Ezra
+          wizard_role: "Infrastructure wizard — Gitea, nginx, hosting"
+          wizard_provider_primary: kimi-coding
+          wizard_model_primary: kimi-k2.5
+          hermes_port: 8081
+          api_port: 8645
+          wizard_home: /root/wizards/ezra
+          hermes_home: /root/.hermes
+          machine_type: vps
+          ollama_available: false
+          # NOTE: Currently DOWN — Telegram key revoked, awaiting propagation.
+
+    # Infrastructure hosts (not wizards, but managed by Ansible)
+    infrastructure:
+      hosts:
+        forge:
+          ansible_host: 143.198.27.163
+          ansible_user: root
+          # Gitea runs on the same box as Ezra
+          gitea_url: https://forge.alexanderwhitestone.com
+          gitea_org: Timmy_Foundation
+
+  vars:
+    # Global variables applied to all hosts
+    gitea_repo_url: "https://forge.alexanderwhitestone.com/Timmy_Foundation/timmy-config.git"
+    gitea_branch: main
+    config_base_path: "{{ gitea_repo_url }}"
+    timmy_log_dir: "~/.local/timmy/fleet-health"
+    request_log_db: "~/.local/timmy/request_log.db"
+
+    # Golden state provider chain — Anthropic is BANNED
+    golden_state_providers:
+      - name: kimi-coding
+        model: kimi-k2.5
+        base_url: "https://api.kimi.com/coding/v1"
+        timeout: 120
+        reason: "Primary — Kimi K2.5 (best value, least friction)"
+      - name: openrouter
+        model: google/gemini-2.5-pro
+        base_url: "https://openrouter.ai/api/v1"
+        api_key_env: OPENROUTER_API_KEY
+        timeout: 120
+        reason: "Fallback — Gemini 2.5 Pro via OpenRouter"
+      - name: ollama
+        model: "gemma4:latest"
+        base_url: "http://localhost:11434/v1"
+        timeout: 180
+        reason: "Terminal fallback — local Ollama (sovereign, no API needed)"
+
+    # Banned providers — hard enforcement
+    banned_providers:
+      - anthropic
+      - claude
+    banned_models_patterns:
+      - "claude-*"
+      - "anthropic/*"
+      - "*sonnet*"
+      - "*opus*"
+      - "*haiku*"

ansible/playbooks/agent_startup.yml

@@ -0,0 +1,98 @@
+---
+# =============================================================================
+# agent_startup.yml — Resurrect Wizards from Checked-in Configs
+# =============================================================================
+# Brings wizards back online using golden state configs.
+# Order: pull config → validate → start agent → verify with request_log
+# =============================================================================
+
+- name: "Agent Startup Sequence"
+  hosts: wizards
+  become: true
+  serial: 1  # One wizard at a time to avoid cascading issues
+
+  tasks:
+    - name: "Pull latest config from upstream"
+      git:
+        repo: "{{ upstream_repo }}"
+        dest: "{{ wizard_home }}/workspace/timmy-config"
+        version: "{{ upstream_branch }}"
+        force: true
+      tags: [pull]
+
+    - name: "Deploy golden state config"
+      include_role:
+        name: golden_state
+      tags: [config]
+
+    - name: "Validate config — no banned providers"
+      shell: |
+        python3 -c "
+        import yaml, sys
+        with open('{{ wizard_home }}/config.yaml') as f:
+            cfg = yaml.safe_load(f)
+        banned = {{ banned_providers }}
+        for p in cfg.get('fallback_providers', []):
+            if p.get('provider', '') in banned:
+                print(f'BANNED: {p[\"provider\"]}', file=sys.stderr)
+                sys.exit(1)
+        model = cfg.get('model', {}).get('provider', '')
+        if model in banned:
+            print(f'BANNED default provider: {model}', file=sys.stderr)
+            sys.exit(1)
+        print('Config validated — no banned providers.')
+        "
+      register: config_valid
+      tags: [validate]
+
+    - name: "Ensure hermes-agent service is running"
+      systemd:
+        name: "hermes-{{ wizard_name | lower }}"
+        state: started
+        enabled: true
+      when: machine_type == 'vps'
+      tags: [start]
+      ignore_errors: true  # Service may not exist yet on all machines
+
+    - name: "Start hermes agent (Mac — launchctl)"
+      shell: |
+        launchctl kickstart -k "ai.hermes.{{ wizard_name | lower }}" 2>/dev/null || \
+        cd {{ wizard_home }} && hermes agent start --daemon 2>&1 | tail -5
+      when: machine_type == 'mac'
+      tags: [start]
+      ignore_errors: true
+
+    - name: "Wait for agent to come online"
+      wait_for:
+        host: 127.0.0.1
+        port: "{{ api_port }}"
+        timeout: 60
+        state: started
+      tags: [verify]
+      ignore_errors: true
+
+    - name: "Verify agent is alive — check request_log for activity"
+      shell: |
+        sleep 10
+        python3 -c "
+        import sqlite3, sys
+        db = sqlite3.connect('{{ request_log_path }}')
+        cursor = db.execute('''
+            SELECT COUNT(*) FROM request_log
+            WHERE agent_name = '{{ wizard_name }}'
+            AND timestamp > datetime('now', '-5 minutes')
+        ''')
+        count = cursor.fetchone()[0]
+        if count > 0:
+            print(f'{{ wizard_name }} is alive — {count} recent inference calls logged.')
+        else:
+            print(f'WARNING: {{ wizard_name }} started but no telemetry yet.')
+        "
+      register: agent_status
+      tags: [verify]
+      ignore_errors: true
+
+    - name: "Report startup status"
+      debug:
+        msg: "{{ wizard_name }}: {{ agent_status.stdout | default('startup attempted') }}"
+      tags: [always]

ansible/playbooks/cron_schedule.yml

@@ -0,0 +1,15 @@
+---
+# =============================================================================
+# cron_schedule.yml — Source-Controlled Cron Jobs
+# =============================================================================
+# All cron jobs are defined in group_vars/wizards.yml.
+# This playbook deploys them. No manual crontab edits allowed.
+# =============================================================================
+
+- name: "Deploy Cron Schedule"
+  hosts: wizards
+  become: true
+
+  roles:
+    - role: cron_manager
+      tags: [cron, schedule]

ansible/playbooks/deadman_switch.yml

@@ -0,0 +1,17 @@
+---
+# =============================================================================
+# deadman_switch.yml — Deploy Deadman Switch to All Wizards
+# =============================================================================
+# The deadman watch already fires and detects dead agents.
+# This playbook wires the ACTION:
+#   - On healthy check: snapshot current config as "last known good"
+#   - On failed check: rollback config to snapshot, restart agent
+# =============================================================================
+
+- name: "Deploy Deadman Switch ACTION"
+  hosts: wizards
+  become: true
+
+  roles:
+    - role: deadman_switch
+      tags: [deadman, recovery]

ansible/playbooks/golden_state.yml

@@ -0,0 +1,30 @@
+---
+# =============================================================================
+# golden_state.yml — Deploy Golden State Config to All Wizards
+# =============================================================================
+# Enforces the golden state provider chain across the fleet.
+# Removes any Anthropic references. Deploys the approved provider chain.
+# =============================================================================
+
+- name: "Deploy Golden State Configuration"
+  hosts: wizards
+  become: true
+
+  roles:
+    - role: golden_state
+      tags: [golden, config]
+
+  post_tasks:
+    - name: "Verify golden state — no banned providers"
+      shell: |
+        grep -rci 'anthropic\|claude-sonnet\|claude-opus\|claude-haiku' \
+          {{ hermes_home }}/config.yaml \
+          {{ wizard_home }}/config.yaml 2>/dev/null || echo "0"
+      register: banned_count
+      changed_when: false
+
+    - name: "Report golden state status"
+      debug:
+        msg: >
+          {{ wizard_name }} golden state: {{ golden_state_providers | map(attribute='name') | list | join(' → ') }}.
+          Banned provider references: {{ banned_count.stdout | trim }}.

ansible/playbooks/request_log.yml

@@ -0,0 +1,15 @@
+---
+# =============================================================================
+# request_log.yml — Deploy Telemetry Table
+# =============================================================================
+# Creates the request_log SQLite table on all machines.
+# Every inference call writes a row. No exceptions. No summarizing.
+# =============================================================================
+
+- name: "Deploy Request Log Telemetry"
+  hosts: wizards
+  become: true
+
+  roles:
+    - role: request_log
+      tags: [telemetry, logging]

ansible/playbooks/site.yml

@@ -0,0 +1,72 @@
+---
+# =============================================================================
+# site.yml — Master Playbook for the Timmy Foundation Fleet
+# =============================================================================
+# This is the ONE playbook that defines the entire fleet state.
+# Run this and every machine converges to golden state.
+#
+# Usage:
+#   ansible-playbook -i inventory/hosts.yml playbooks/site.yml
+#   ansible-playbook -i inventory/hosts.yml playbooks/site.yml --limit bezalel
+#   ansible-playbook -i inventory/hosts.yml playbooks/site.yml --check --diff
+# =============================================================================
+
+- name: "Timmy Foundation Fleet — Full Convergence"
+  hosts: wizards
+  become: true
+
+  pre_tasks:
+    - name: "Validate no banned providers in golden state"
+      assert:
+        that:
+          - "item.name not in banned_providers"
+        fail_msg: "BANNED PROVIDER DETECTED: {{ item.name }} — Anthropic is permanently banned."
+        quiet: true
+      loop: "{{ golden_state_providers }}"
+      tags: [always]
+
+    - name: "Display target wizard"
+      debug:
+        msg: "Deploying to {{ wizard_name }} ({{ wizard_role }}) on {{ ansible_host }}"
+      tags: [always]
+
+  roles:
+    - role: wizard_base
+      tags: [base, setup]
+
+    - role: golden_state
+      tags: [golden, config]
+
+    - role: deadman_switch
+      tags: [deadman, recovery]
+
+    - role: request_log
+      tags: [telemetry, logging]
+
+    - role: cron_manager
+      tags: [cron, schedule]
+
+  post_tasks:
+    - name: "Final validation — scan for banned providers"
+      shell: |
+        grep -ri 'anthropic\|claude-sonnet\|claude-opus\|claude-haiku' \
+          {{ hermes_home }}/config.yaml \
+          {{ wizard_home }}/config.yaml \
+          {{ thin_config_path }} 2>/dev/null || true
+      register: banned_scan
+      changed_when: false
+      tags: [validation]
+
+    - name: "FAIL if banned providers found in deployed config"
+      fail:
+        msg: |
+          BANNED PROVIDER DETECTED IN DEPLOYED CONFIG:
+          {{ banned_scan.stdout }}
+          Anthropic is permanently banned. Fix the config and re-deploy.
+      when: banned_scan.stdout | length > 0
+      tags: [validation]
+
+    - name: "Deployment complete"
+      debug:
+        msg: "{{ wizard_name }} converged to golden state. Provider chain: {{ golden_state_providers | map(attribute='name') | list | join(' → ') }}"
+      tags: [always]

ansible/roles/cron_manager/tasks/main.yml

@@ -0,0 +1,55 @@
+---
+# =============================================================================
+# cron_manager/tasks — Source-Controlled Cron Jobs
+# =============================================================================
+# All cron jobs are defined in group_vars/wizards.yml.
+# No manual crontab edits. This is the only way to manage cron.
+# =============================================================================
+
+- name: "Deploy managed cron jobs"
+  cron:
+    name: "{{ item.name }}"
+    job: "{{ item.job }}"
+    minute: "{{ item.minute | default('*') }}"
+    hour: "{{ item.hour | default('*') }}"
+    day: "{{ item.day | default('*') }}"
+    month: "{{ item.month | default('*') }}"
+    weekday: "{{ item.weekday | default('*') }}"
+    state: "{{ 'present' if item.enabled else 'absent' }}"
+    user: "{{ ansible_user | default('root') }}"
+  loop: "{{ cron_jobs }}"
+  when: cron_jobs is defined
+
+- name: "Deploy deadman switch cron (fallback if systemd timer unavailable)"
+  cron:
+    name: "Deadman switch — {{ wizard_name }}"
+    job: "{{ wizard_home }}/deadman_action.sh >> {{ timmy_log_dir }}/deadman-{{ wizard_name }}.log 2>&1"
+    minute: "*/5"
+    hour: "*"
+    state: present
+    user: "{{ ansible_user | default('root') }}"
+  when: deadman_enabled and machine_type != 'vps'
+  # VPS machines use systemd timers instead
+
+- name: "Remove legacy cron jobs (cleanup)"
+  cron:
+    name: "{{ item }}"
+    state: absent
+    user: "{{ ansible_user | default('root') }}"
+  loop:
+    - "legacy-deadman-watch"
+    - "old-health-check"
+    - "backup-deadman"
+  ignore_errors: true
+
+- name: "List active cron jobs"
+  shell: "crontab -l 2>/dev/null | grep -v '^#' | grep -v '^$' || echo 'No cron jobs found.'"
+  register: active_crons
+  changed_when: false
+
+- name: "Report cron status"
+  debug:
+    msg: |
+      {{ wizard_name }} cron jobs deployed.
+      Active:
+      {{ active_crons.stdout }}

ansible/roles/deadman_switch/tasks/main.yml

@@ -0,0 +1,70 @@
+---
+# =============================================================================
+# deadman_switch/tasks — Wire the Deadman Switch ACTION
+# =============================================================================
+# The watch fires. This makes it DO something:
+#   - On healthy check: snapshot current config as "last known good"
+#   - On failed check: rollback to last known good, restart agent
+# =============================================================================
+
+- name: "Create snapshot directory"
+  file:
+    path: "{{ deadman_snapshot_dir }}"
+    state: directory
+    mode: "0755"
+
+- name: "Deploy deadman switch script"
+  template:
+    src: deadman_action.sh.j2
+    dest: "{{ wizard_home }}/deadman_action.sh"
+    mode: "0755"
+
+- name: "Deploy deadman systemd service"
+  template:
+    src: deadman_switch.service.j2
+    dest: "/etc/systemd/system/deadman-{{ wizard_name | lower }}.service"
+    mode: "0644"
+  when: machine_type == 'vps'
+  notify: "Enable deadman service"
+
+- name: "Deploy deadman systemd timer"
+  template:
+    src: deadman_switch.timer.j2
+    dest: "/etc/systemd/system/deadman-{{ wizard_name | lower }}.timer"
+    mode: "0644"
+  when: machine_type == 'vps'
+  notify: "Enable deadman timer"
+
+- name: "Deploy deadman launchd plist (Mac)"
+  template:
+    src: deadman_switch.plist.j2
+    dest: "{{ ansible_env.HOME }}/Library/LaunchAgents/com.timmy.deadman.{{ wizard_name | lower }}.plist"
+    mode: "0644"
+  when: machine_type == 'mac'
+  notify: "Load deadman plist"
+
+- name: "Take initial config snapshot"
+  copy:
+    src: "{{ wizard_home }}/config.yaml"
+    dest: "{{ deadman_snapshot_dir }}/config.yaml.known_good"
+    remote_src: true
+    mode: "0444"
+  ignore_errors: true
+
+handlers:
+  - name: "Enable deadman service"
+    systemd:
+      name: "deadman-{{ wizard_name | lower }}.service"
+      daemon_reload: true
+      enabled: true
+
+  - name: "Enable deadman timer"
+    systemd:
+      name: "deadman-{{ wizard_name | lower }}.timer"
+      daemon_reload: true
+      enabled: true
+      state: started
+
+  - name: "Load deadman plist"
+    shell: "launchctl load {{ ansible_env.HOME }}/Library/LaunchAgents/com.timmy.deadman.{{ wizard_name | lower }}.plist"
+    ignore_errors: true

ansible/roles/deadman_switch/templates/deadman_action.sh.j2

@@ -0,0 +1,153 @@
+#!/usr/bin/env bash
+# =============================================================================
+# Deadman Switch ACTION — {{ wizard_name }}
+# =============================================================================
+# Generated by Ansible on {{ ansible_date_time.iso8601 }}
+# DO NOT EDIT MANUALLY.
+#
+# On healthy check: snapshot current config as "last known good"
+# On failed check: rollback config to last known good, restart agent
+# =============================================================================
+
+set -euo pipefail
+
+WIZARD_NAME="{{ wizard_name }}"
+WIZARD_HOME="{{ wizard_home }}"
+CONFIG_FILE="{{ wizard_home }}/config.yaml"
+SNAPSHOT_DIR="{{ deadman_snapshot_dir }}"
+SNAPSHOT_FILE="${SNAPSHOT_DIR}/config.yaml.known_good"
+REQUEST_LOG_DB="{{ request_log_path }}"
+LOG_DIR="{{ timmy_log_dir }}"
+LOG_FILE="${LOG_DIR}/deadman-${WIZARD_NAME}.log"
+MAX_SNAPSHOTS={{ deadman_max_snapshots }}
+RESTART_COOLDOWN={{ deadman_restart_cooldown }}
+MAX_RESTART_ATTEMPTS={{ deadman_max_restart_attempts }}
+COOLDOWN_FILE="${LOG_DIR}/deadman_cooldown_${WIZARD_NAME}"
+SERVICE_NAME="hermes-{{ wizard_name | lower }}"
+
+# Ensure directories exist
+mkdir -p "${SNAPSHOT_DIR}" "${LOG_DIR}"
+
+log() {
+    echo "[$(date -u +%Y-%m-%dT%H:%M:%SZ)] [deadman] [${WIZARD_NAME}] $*" >> "${LOG_FILE}"
+    echo "[deadman] [${WIZARD_NAME}] $*"
+}
+
+log_telemetry() {
+    local status="$1"
+    local message="$2"
+    if [ -f "${REQUEST_LOG_DB}" ]; then
+        sqlite3 "${REQUEST_LOG_DB}" "INSERT INTO request_log (timestamp, agent_name, provider, model, endpoint, status, error_message) VALUES (datetime('now'), '${WIZARD_NAME}', 'deadman_switch', 'N/A', 'health_check', '${status}', '${message}');" 2>/dev/null || true
+    fi
+}
+
+snapshot_config() {
+    if [ -f "${CONFIG_FILE}" ]; then
+        cp "${CONFIG_FILE}" "${SNAPSHOT_FILE}"
+        # Keep rolling history
+        cp "${CONFIG_FILE}" "${SNAPSHOT_DIR}/config.yaml.$(date +%s)"
+        # Prune old snapshots
+        ls -t "${SNAPSHOT_DIR}"/config.yaml.[0-9]* 2>/dev/null | tail -n +$((MAX_SNAPSHOTS + 1)) | xargs rm -f 2>/dev/null
+        log "Config snapshot saved."
+    fi
+}
+
+rollback_config() {
+    if [ -f "${SNAPSHOT_FILE}" ]; then
+        log "Rolling back config to last known good..."
+        cp "${SNAPSHOT_FILE}" "${CONFIG_FILE}"
+        log "Config rolled back."
+        log_telemetry "fallback" "Config rolled back to last known good by deadman switch"
+    else
+        log "ERROR: No known good snapshot found. Pulling from upstream..."
+        cd "${WIZARD_HOME}/workspace/timmy-config" 2>/dev/null && \
+            git pull --ff-only origin {{ upstream_branch }} 2>/dev/null && \
+            cp "wizards/{{ wizard_name | lower }}/config.yaml" "${CONFIG_FILE}" && \
+            log "Config restored from upstream." || \
+            log "CRITICAL: Cannot restore config from any source."
+    fi
+}
+
+restart_agent() {
+    # Check cooldown
+    if [ -f "${COOLDOWN_FILE}" ]; then
+        local last_restart
+        last_restart=$(cat "${COOLDOWN_FILE}")
+        local now
+        now=$(date +%s)
+        local elapsed=$((now - last_restart))
+        if [ "${elapsed}" -lt "${RESTART_COOLDOWN}" ]; then
+            log "Restart cooldown active (${elapsed}s / ${RESTART_COOLDOWN}s). Skipping."
+            return 1
+        fi
+    fi
+
+    log "Restarting ${SERVICE_NAME}..."
+    date +%s > "${COOLDOWN_FILE}"
+
+{% if machine_type == 'vps' %}
+    systemctl restart "${SERVICE_NAME}" 2>/dev/null && \
+        log "Agent restarted via systemd." || \
+        log "ERROR: systemd restart failed."
+{% else %}
+    launchctl kickstart -k "ai.hermes.{{ wizard_name | lower }}" 2>/dev/null && \
+        log "Agent restarted via launchctl." || \
+        (cd "${WIZARD_HOME}" && hermes agent start --daemon 2>/dev/null && \
+        log "Agent restarted via hermes CLI.") || \
+        log "ERROR: All restart methods failed."
+{% endif %}
+
+    log_telemetry "success" "Agent restarted by deadman switch"
+}
+
+# --- Health Check ---
+check_health() {
+    # Check 1: Is the agent process running?
+{% if machine_type == 'vps' %}
+    if ! systemctl is-active --quiet "${SERVICE_NAME}" 2>/dev/null; then
+        if ! pgrep -f "hermes" > /dev/null 2>/dev/null; then
+            log "FAIL: Agent process not running."
+            return 1
+        fi
+    fi
+{% else %}
+    if ! pgrep -f "hermes" > /dev/null 2>/dev/null; then
+        log "FAIL: Agent process not running."
+        return 1
+    fi
+{% endif %}
+
+    # Check 2: Is the API port responding?
+    if ! timeout 10 bash -c "echo > /dev/tcp/127.0.0.1/{{ api_port }}" 2>/dev/null; then
+        log "FAIL: API port {{ api_port }} not responding."
+        return 1
+    fi
+
+    # Check 3: Does the config contain banned providers?
+    if grep -qi 'anthropic\|claude-sonnet\|claude-opus\|claude-haiku' "${CONFIG_FILE}" 2>/dev/null; then
+        log "FAIL: Config contains banned provider (Anthropic). Rolling back."
+        return 1
+    fi
+
+    return 0
+}
+
+# --- Main ---
+main() {
+    log "Health check starting..."
+
+    if check_health; then
+        log "HEALTHY — snapshotting config."
+        snapshot_config
+        log_telemetry "success" "Health check passed"
+    else
+        log "UNHEALTHY — initiating recovery."
+        log_telemetry "error" "Health check failed — initiating rollback"
+        rollback_config
+        restart_agent
+    fi
+
+    log "Health check complete."
+}
+
+main "$@"

ansible/roles/deadman_switch/templates/deadman_switch.plist.j2

@@ -0,0 +1,22 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!DOCTYPE plist PUBLIC "-//Apple//DTD PLIST 1.0//EN" "http://www.apple.com/DTDs/PropertyList-1.0.dtd">
+<!-- Deadman Switch — {{ wizard_name }}. Generated by Ansible. DO NOT EDIT MANUALLY. -->
+<plist version="1.0">
+<dict>
+    <key>Label</key>
+    <string>com.timmy.deadman.{{ wizard_name | lower }}</string>
+    <key>ProgramArguments</key>
+    <array>
+        <string>/bin/bash</string>
+        <string>{{ wizard_home }}/deadman_action.sh</string>
+    </array>
+    <key>StartInterval</key>
+    <integer>{{ deadman_check_interval }}</integer>
+    <key>RunAtLoad</key>
+    <true/>
+    <key>StandardOutPath</key>
+    <string>{{ timmy_log_dir }}/deadman-{{ wizard_name }}.log</string>
+    <key>StandardErrorPath</key>
+    <string>{{ timmy_log_dir }}/deadman-{{ wizard_name }}.log</string>
+</dict>
+</plist>

ansible/roles/deadman_switch/templates/deadman_switch.service.j2

@@ -0,0 +1,16 @@
+# Deadman Switch — {{ wizard_name }}
+# Generated by Ansible. DO NOT EDIT MANUALLY.
+
+[Unit]
+Description=Deadman Switch for {{ wizard_name }} wizard
+After=network.target
+
+[Service]
+Type=oneshot
+ExecStart={{ wizard_home }}/deadman_action.sh
+User={{ ansible_user | default('root') }}
+StandardOutput=append:{{ timmy_log_dir }}/deadman-{{ wizard_name }}.log
+StandardError=append:{{ timmy_log_dir }}/deadman-{{ wizard_name }}.log
+
+[Install]
+WantedBy=multi-user.target

ansible/roles/deadman_switch/templates/deadman_switch.timer.j2

@@ -0,0 +1,14 @@
+# Deadman Switch Timer — {{ wizard_name }}
+# Generated by Ansible. DO NOT EDIT MANUALLY.
+# Runs every {{ deadman_check_interval // 60 }} minutes.
+
+[Unit]
+Description=Deadman Switch Timer for {{ wizard_name }} wizard
+
+[Timer]
+OnBootSec=60
+OnUnitActiveSec={{ deadman_check_interval }}s
+AccuracySec=30s
+
+[Install]
+WantedBy=timers.target

ansible/roles/golden_state/defaults/main.yml

@@ -0,0 +1,6 @@
+---
+# golden_state defaults
+# The golden_state_providers list is defined in group_vars/wizards.yml
+# and inventory/hosts.yml (global vars).
+golden_state_enforce: true
+golden_state_backup_before_deploy: true

ansible/roles/golden_state/tasks/main.yml

@@ -0,0 +1,46 @@
+---
+# =============================================================================
+# golden_state/tasks — Deploy and enforce golden state provider chain
+# =============================================================================
+
+- name: "Backup current config before golden state deploy"
+  copy:
+    src: "{{ wizard_home }}/config.yaml"
+    dest: "{{ wizard_home }}/config.yaml.pre-golden-{{ ansible_date_time.epoch }}"
+    remote_src: true
+  when: golden_state_backup_before_deploy
+  ignore_errors: true
+
+- name: "Deploy golden state wizard config"
+  template:
+    src: "../../wizard_base/templates/wizard_config.yaml.j2"
+    dest: "{{ wizard_home }}/config.yaml"
+    mode: "0644"
+    backup: true
+  notify:
+    - "Restart hermes agent (systemd)"
+    - "Restart hermes agent (launchctl)"
+
+- name: "Scan for banned providers in all config files"
+  shell: |
+    FOUND=0
+    for f in {{ wizard_home }}/config.yaml {{ hermes_home }}/config.yaml; do
+      if [ -f "$f" ]; then
+        if grep -qi 'anthropic\|claude-sonnet\|claude-opus\|claude-haiku' "$f"; then
+          echo "BANNED PROVIDER in $f:"
+          grep -ni 'anthropic\|claude-sonnet\|claude-opus\|claude-haiku' "$f"
+          FOUND=1
+        fi
+      fi
+    done
+    exit $FOUND
+  register: provider_scan
+  changed_when: false
+  failed_when: provider_scan.rc != 0 and provider_ban_enforcement == 'strict'
+
+- name: "Report golden state deployment"
+  debug:
+    msg: >
+      {{ wizard_name }} golden state deployed.
+      Provider chain: {{ golden_state_providers | map(attribute='name') | list | join(' → ') }}.
+      Banned provider scan: {{ 'CLEAN' if provider_scan.rc == 0 else 'VIOLATIONS FOUND' }}.

ansible/roles/request_log/files/request_log_schema.sql

@@ -0,0 +1,64 @@
+-- =============================================================================
+-- request_log — Inference Telemetry Table
+-- =============================================================================
+-- Every agent writes to this table BEFORE and AFTER every inference call.
+-- No exceptions. No summarizing. No describing what you would log.
+-- Actually write the row.
+--
+-- Source: KT Bezalel Architecture Session 2026-04-08
+-- =============================================================================
+
+CREATE TABLE IF NOT EXISTS request_log (
+    id INTEGER PRIMARY KEY AUTOINCREMENT,
+    timestamp TEXT NOT NULL DEFAULT (datetime('now')),
+    agent_name TEXT NOT NULL,
+    provider TEXT NOT NULL,
+    model TEXT NOT NULL,
+    endpoint TEXT NOT NULL,
+    tokens_in INTEGER,
+    tokens_out INTEGER,
+    latency_ms INTEGER,
+    status TEXT NOT NULL,  -- 'success', 'error', 'timeout', 'fallback'
+    error_message TEXT
+);
+
+-- Index for common queries
+CREATE INDEX IF NOT EXISTS idx_request_log_agent
+    ON request_log (agent_name, timestamp);
+
+CREATE INDEX IF NOT EXISTS idx_request_log_provider
+    ON request_log (provider, timestamp);
+
+CREATE INDEX IF NOT EXISTS idx_request_log_status
+    ON request_log (status, timestamp);
+
+-- View: recent activity per agent (last hour)
+CREATE VIEW IF NOT EXISTS v_recent_activity AS
+    SELECT
+        agent_name,
+        provider,
+        model,
+        status,
+        COUNT(*) as call_count,
+        AVG(latency_ms) as avg_latency_ms,
+        SUM(tokens_in) as total_tokens_in,
+        SUM(tokens_out) as total_tokens_out
+    FROM request_log
+    WHERE timestamp > datetime('now', '-1 hour')
+    GROUP BY agent_name, provider, model, status;
+
+-- View: provider reliability (last 24 hours)
+CREATE VIEW IF NOT EXISTS v_provider_reliability AS
+    SELECT
+        provider,
+        model,
+        COUNT(*) as total_calls,
+        SUM(CASE WHEN status = 'success' THEN 1 ELSE 0 END) as successes,
+        SUM(CASE WHEN status = 'error' THEN 1 ELSE 0 END) as errors,
+        SUM(CASE WHEN status = 'timeout' THEN 1 ELSE 0 END) as timeouts,
+        SUM(CASE WHEN status = 'fallback' THEN 1 ELSE 0 END) as fallbacks,
+        ROUND(100.0 * SUM(CASE WHEN status = 'success' THEN 1 ELSE 0 END) / COUNT(*), 1) as success_rate,
+        AVG(latency_ms) as avg_latency_ms
+    FROM request_log
+    WHERE timestamp > datetime('now', '-24 hours')
+    GROUP BY provider, model;

ansible/roles/request_log/tasks/main.yml

@@ -0,0 +1,50 @@
+---
+# =============================================================================
+# request_log/tasks — Deploy Telemetry Table
+# =============================================================================
+# "This is non-negotiable infrastructure. Without it, we cannot verify
+# if any agent actually executed what it claims."
+# — KT Bezalel 2026-04-08
+# =============================================================================
+
+- name: "Create telemetry directory"
+  file:
+    path: "{{ request_log_path | dirname }}"
+    state: directory
+    mode: "0755"
+
+- name: "Deploy request_log schema"
+  copy:
+    src: request_log_schema.sql
+    dest: "{{ wizard_home }}/request_log_schema.sql"
+    mode: "0644"
+
+- name: "Initialize request_log database"
+  shell: |
+    sqlite3 "{{ request_log_path }}" < "{{ wizard_home }}/request_log_schema.sql"
+  args:
+    creates: "{{ request_log_path }}"
+
+- name: "Verify request_log table exists"
+  shell: |
+    sqlite3 "{{ request_log_path }}" ".tables" | grep -q "request_log"
+  register: table_check
+  changed_when: false
+
+- name: "Verify request_log schema matches"
+  shell: |
+    sqlite3 "{{ request_log_path }}" ".schema request_log" | grep -q "agent_name"
+  register: schema_check
+  changed_when: false
+
+- name: "Set permissions on request_log database"
+  file:
+    path: "{{ request_log_path }}"
+    mode: "0644"
+
+- name: "Report request_log status"
+  debug:
+    msg: >
+      {{ wizard_name }} request_log: {{ request_log_path }}
+      — table exists: {{ table_check.rc == 0 }}
+      — schema valid: {{ schema_check.rc == 0 }}

ansible/roles/wizard_base/defaults/main.yml

@@ -0,0 +1,6 @@
+---
+# wizard_base defaults
+wizard_user: "{{ ansible_user | default('root') }}"
+wizard_group: "{{ ansible_user | default('root') }}"
+timmy_base_dir: "~/.local/timmy"
+timmy_config_repo: "https://forge.alexanderwhitestone.com/Timmy_Foundation/timmy-config.git"

ansible/roles/wizard_base/handlers/main.yml

@@ -0,0 +1,11 @@
+---
+- name: "Restart hermes agent (systemd)"
+  systemd:
+    name: "hermes-{{ wizard_name | lower }}"
+    state: restarted
+  when: machine_type == 'vps'
+
+- name: "Restart hermes agent (launchctl)"
+  shell: "launchctl kickstart -k ai.hermes.{{ wizard_name | lower }}"
+  when: machine_type == 'mac'
+  ignore_errors: true

ansible/roles/wizard_base/tasks/main.yml

@@ -0,0 +1,69 @@
+---
+# =============================================================================
+# wizard_base/tasks — Common wizard setup
+# =============================================================================
+
+- name: "Create wizard directories"
+  file:
+    path: "{{ item }}"
+    state: directory
+    mode: "0755"
+  loop:
+    - "{{ wizard_home }}"
+    - "{{ wizard_home }}/workspace"
+    - "{{ hermes_home }}"
+    - "{{ hermes_home }}/bin"
+    - "{{ hermes_home }}/skins"
+    - "{{ hermes_home }}/playbooks"
+    - "{{ hermes_home }}/memories"
+    - "~/.local/timmy"
+    - "~/.local/timmy/fleet-health"
+    - "~/.local/timmy/snapshots"
+    - "~/.timmy"
+
+- name: "Clone/update timmy-config"
+  git:
+    repo: "{{ upstream_repo }}"
+    dest: "{{ wizard_home }}/workspace/timmy-config"
+    version: "{{ upstream_branch }}"
+    force: false
+    update: true
+  ignore_errors: true  # May fail on first run if no SSH key
+
+- name: "Deploy SOUL.md"
+  copy:
+    src: "{{ wizard_home }}/workspace/timmy-config/SOUL.md"
+    dest: "~/.timmy/SOUL.md"
+    remote_src: true
+    mode: "0644"
+  ignore_errors: true
+
+- name: "Deploy thin config (immutable pointer to upstream)"
+  template:
+    src: thin_config.yml.j2
+    dest: "{{ thin_config_path }}"
+    mode: "{{ thin_config_mode }}"
+  tags: [thin_config]
+
+- name: "Ensure Python3 and pip are available"
+  package:
+    name:
+      - python3
+      - python3-pip
+    state: present
+  when: machine_type == 'vps'
+  ignore_errors: true
+
+- name: "Ensure PyYAML is installed (for config validation)"
+  pip:
+    name: pyyaml
+    state: present
+  when: machine_type == 'vps'
+  ignore_errors: true
+
+- name: "Create Ansible log directory"
+  file:
+    path: /var/log/ansible
+    state: directory
+    mode: "0755"
+  ignore_errors: true

ansible/roles/wizard_base/templates/thin_config.yml.j2

@@ -0,0 +1,41 @@
+# =============================================================================
+# Thin Config — {{ wizard_name }}
+# =============================================================================
+# THIS FILE IS READ-ONLY. Agents CANNOT modify it.
+# It contains only pointers to upstream. The actual config lives in Gitea.
+#
+# Agent wakes up → pulls config from upstream → loads → runs.
+# If anything tries to mutate this → fails gracefully → pulls fresh on restart.
+#
+# Only way to permanently change config: commit to Gitea, merge PR, Ansible deploys.
+#
+# Generated by Ansible on {{ ansible_date_time.iso8601 }}
+# DO NOT EDIT MANUALLY.
+# =============================================================================
+
+identity:
+  wizard_name: "{{ wizard_name }}"
+  wizard_role: "{{ wizard_role }}"
+  machine: "{{ inventory_hostname }}"
+
+upstream:
+  repo: "{{ upstream_repo }}"
+  branch: "{{ upstream_branch }}"
+  config_path: "wizards/{{ wizard_name | lower }}/config.yaml"
+  pull_on_wake: {{ config_pull_on_wake | lower }}
+
+recovery:
+  deadman_enabled: {{ deadman_enabled | lower }}
+  snapshot_dir: "{{ deadman_snapshot_dir }}"
+  restart_cooldown: {{ deadman_restart_cooldown }}
+  max_restart_attempts: {{ deadman_max_restart_attempts }}
+  escalation_channel: "{{ deadman_escalation_channel }}"
+
+telemetry:
+  request_log_path: "{{ request_log_path }}"
+  request_log_enabled: {{ request_log_enabled | lower }}
+
+local_overrides:
+  # Runtime overrides go here. They are EPHEMERAL — not persisted across restarts.
+  # On restart, this section is reset to empty.
+  {}

ansible/roles/wizard_base/templates/wizard_config.yaml.j2

@@ -0,0 +1,115 @@
+# =============================================================================
+# {{ wizard_name }} — Wizard Configuration (Golden State)
+# =============================================================================
+# Generated by Ansible on {{ ansible_date_time.iso8601 }}
+# DO NOT EDIT MANUALLY. Changes go through Gitea PR → Ansible deploy.
+#
+# Provider chain: {{ golden_state_providers | map(attribute='name') | list | join(' → ') }}
+# Anthropic is PERMANENTLY BANNED.
+# =============================================================================
+
+model:
+  default: {{ wizard_model_primary }}
+  provider: {{ wizard_provider_primary }}
+  context_length: 65536
+  base_url: {{ golden_state_providers[0].base_url }}
+
+toolsets:
+  - all
+
+fallback_providers:
+{% for provider in golden_state_providers %}
+  - provider: {{ provider.name }}
+    model: {{ provider.model }}
+{% if provider.base_url is defined %}
+    base_url: {{ provider.base_url }}
+{% endif %}
+{% if provider.api_key_env is defined %}
+    api_key_env: {{ provider.api_key_env }}
+{% endif %}
+    timeout: {{ provider.timeout }}
+    reason: "{{ provider.reason }}"
+{% endfor %}
+
+agent:
+  max_turns: {{ agent_max_turns }}
+  reasoning_effort: {{ agent_reasoning_effort }}
+  verbose: {{ agent_verbose | lower }}
+
+terminal:
+  backend: local
+  cwd: .
+  timeout: 180
+  persistent_shell: true
+
+browser:
+  inactivity_timeout: 120
+  command_timeout: 30
+  record_sessions: false
+
+display:
+  compact: false
+  personality: ''
+  resume_display: full
+  busy_input_mode: interrupt
+  bell_on_complete: false
+  show_reasoning: false
+  streaming: false
+  show_cost: false
+  tool_progress: all
+
+memory:
+  memory_enabled: true
+  user_profile_enabled: true
+  memory_char_limit: 2200
+  user_char_limit: 1375
+  nudge_interval: 10
+  flush_min_turns: 6
+
+approvals:
+  mode: {{ agent_approval_mode }}
+
+security:
+  redact_secrets: true
+  tirith_enabled: false
+
+platforms:
+  api_server:
+    enabled: true
+    extra:
+      host: 127.0.0.1
+      port: {{ api_port }}
+
+session_reset:
+  mode: none
+  idle_minutes: 0
+
+skills:
+  creation_nudge_interval: 15
+
+system_prompt_suffix: |
+  You are {{ wizard_name }}, {{ wizard_role }}.
+  Your soul is defined in SOUL.md — read it, live it.
+  Hermes is your harness.
+  {{ golden_state_providers[0].name }} is your primary provider.
+  Refusal over fabrication. If you do not know, say so.
+  Sovereignty and service always.
+
+providers:
+{% for provider in golden_state_providers %}
+  {{ provider.name }}:
+    base_url: {{ provider.base_url }}
+    timeout: {{ provider.timeout | default(60) }}
+{% if provider.name == 'kimi-coding' %}
+    max_retries: 3
+{% endif %}
+{% endfor %}
+
+# =============================================================================
+# BANNED PROVIDERS — DO NOT ADD
+# =============================================================================
+# The following providers are PERMANENTLY BANNED:
+# - anthropic (any model: claude-sonnet, claude-opus, claude-haiku)
+# Enforcement: pre-commit hook, linter, Ansible validation, this comment.
+# Adding any banned provider will cause Ansible deployment to FAIL.
+# =============================================================================

ansible/scripts/deploy_on_webhook.sh

@@ -0,0 +1,75 @@
+#!/usr/bin/env bash
+# =============================================================================
+# Gitea Webhook Handler — Trigger Ansible Deploy on Merge
+# =============================================================================
+# This script is called by the Gitea webhook when a PR is merged
+# to the main branch of timmy-config.
+#
+# Setup:
+#   1. Add webhook in Gitea: Settings → Webhooks → Add Webhook
+#   2. URL: http://localhost:9000/hooks/deploy-timmy-config
+#   3. Events: Pull Request (merged only)
+#   4. Secret: <configured in Gitea>
+#
+# This script runs ansible-pull to update the local machine.
+# For fleet-wide deploys, each machine runs ansible-pull independently.
+# =============================================================================
+
+set -euo pipefail
+
+REPO="https://forge.alexanderwhitestone.com/Timmy_Foundation/timmy-config.git"
+BRANCH="main"
+ANSIBLE_DIR="ansible"
+LOG_FILE="/var/log/ansible/webhook-deploy.log"
+LOCK_FILE="/tmp/ansible-deploy.lock"
+
+log() {
+    echo "[$(date -u +%Y-%m-%dT%H:%M:%SZ)] [webhook] $*" | tee -a "${LOG_FILE}"
+}
+
+# Prevent concurrent deploys
+if [ -f "${LOCK_FILE}" ]; then
+    LOCK_AGE=$(( $(date +%s) - $(stat -c %Y "${LOCK_FILE}" 2>/dev/null || echo 0) ))
+    if [ "${LOCK_AGE}" -lt 300 ]; then
+        log "Deploy already in progress (lock age: ${LOCK_AGE}s). Skipping."
+        exit 0
+    else
+        log "Stale lock file (${LOCK_AGE}s old). Removing."
+        rm -f "${LOCK_FILE}"
+    fi
+fi
+
+trap 'rm -f "${LOCK_FILE}"' EXIT
+touch "${LOCK_FILE}"
+
+log "Webhook triggered. Starting ansible-pull..."
+
+# Pull latest config
+cd /tmp
+rm -rf timmy-config-deploy
+git clone --depth 1 --branch "${BRANCH}" "${REPO}" timmy-config-deploy 2>&1 | tee -a "${LOG_FILE}"
+
+cd timmy-config-deploy/${ANSIBLE_DIR}
+
+# Run Ansible against localhost
+log "Running Ansible playbook..."
+ansible-playbook \
+    -i inventory/hosts.yml \
+    playbooks/site.yml \
+    --limit "$(hostname)" \
+    --diff \
+    2>&1 | tee -a "${LOG_FILE}"
+
+RESULT=$?
+
+if [ ${RESULT} -eq 0 ]; then
+    log "Deploy successful."
+else
+    log "ERROR: Deploy failed with exit code ${RESULT}."
+fi
+
+# Cleanup
+rm -rf /tmp/timmy-config-deploy
+
+log "Webhook handler complete."
+exit ${RESULT}

ansible/scripts/validate_config.py

@@ -0,0 +1,155 @@
+#!/usr/bin/env python3
+"""
+Config Validator — The Timmy Foundation
+Validates wizard configs against golden state rules.
+Run before any config deploy to catch violations early.
+
+Usage:
+    python3 validate_config.py <config_file>
+    python3 validate_config.py --all  # Validate all wizard configs
+
+Exit codes:
+    0 — All validations passed
+    1 — Validation errors found
+    2 — File not found or parse error
+"""
+
+import sys
+import os
+import yaml
+import fnmatch
+from pathlib import Path
+
+# === BANNED PROVIDERS — HARD POLICY ===
+BANNED_PROVIDERS = {"anthropic", "claude"}
+BANNED_MODEL_PATTERNS = [
+    "claude-*",
+    "anthropic/*",
+    "*sonnet*",
+    "*opus*",
+    "*haiku*",
+]
+
+# === REQUIRED FIELDS ===
+REQUIRED_FIELDS = {
+    "model": ["default", "provider"],
+    "fallback_providers": None,  # Must exist as a list
+}
+
+
+def is_banned_model(model_name: str) -> bool:
+    """Check if a model name matches any banned pattern."""
+    model_lower = model_name.lower()
+    for pattern in BANNED_MODEL_PATTERNS:
+        if fnmatch.fnmatch(model_lower, pattern):
+            return True
+    return False
+
+
+def validate_config(config_path: str) -> list[str]:
+    """Validate a wizard config file. Returns list of error strings."""
+    errors = []
+
+    try:
+        with open(config_path) as f:
+            cfg = yaml.safe_load(f)
+    except FileNotFoundError:
+        return [f"File not found: {config_path}"]
+    except yaml.YAMLError as e:
+        return [f"YAML parse error: {e}"]
+
+    if not cfg:
+        return ["Config file is empty"]
+
+    # Check required fields
+    for section, fields in REQUIRED_FIELDS.items():
+        if section not in cfg:
+            errors.append(f"Missing required section: {section}")
+        elif fields:
+            for field in fields:
+                if field not in cfg[section]:
+                    errors.append(f"Missing required field: {section}.{field}")
+
+    # Check default provider
+    default_provider = cfg.get("model", {}).get("provider", "")
+    if default_provider.lower() in BANNED_PROVIDERS:
+        errors.append(f"BANNED default provider: {default_provider}")
+
+    default_model = cfg.get("model", {}).get("default", "")
+    if is_banned_model(default_model):
+        errors.append(f"BANNED default model: {default_model}")
+
+    # Check fallback providers
+    for i, fb in enumerate(cfg.get("fallback_providers", [])):
+        provider = fb.get("provider", "")
+        model = fb.get("model", "")
+
+        if provider.lower() in BANNED_PROVIDERS:
+            errors.append(f"BANNED fallback provider [{i}]: {provider}")
+
+        if is_banned_model(model):
+            errors.append(f"BANNED fallback model [{i}]: {model}")
+
+    # Check providers section
+    for name, provider_cfg in cfg.get("providers", {}).items():
+        if name.lower() in BANNED_PROVIDERS:
+            errors.append(f"BANNED provider in providers section: {name}")
+
+        base_url = str(provider_cfg.get("base_url", ""))
+        if "anthropic" in base_url.lower():
+            errors.append(f"BANNED URL in provider {name}: {base_url}")
+
+    # Check system prompt for banned references
+    prompt = cfg.get("system_prompt_suffix", "")
+    if isinstance(prompt, str):
+        for banned in BANNED_PROVIDERS:
+            if banned in prompt.lower():
+                errors.append(f"BANNED provider referenced in system_prompt_suffix: {banned}")
+
+    return errors
+
+
+def main():
+    if len(sys.argv) < 2:
+        print(f"Usage: {sys.argv[0]} <config_file> [--all]")
+        sys.exit(2)
+
+    if sys.argv[1] == "--all":
+        # Validate all wizard configs in the repo
+        repo_root = Path(__file__).parent.parent.parent
+        wizard_dir = repo_root / "wizards"
+        all_errors = {}
+
+        for wizard_path in sorted(wizard_dir.iterdir()):
+            config_file = wizard_path / "config.yaml"
+            if config_file.exists():
+                errors = validate_config(str(config_file))
+                if errors:
+                    all_errors[wizard_path.name] = errors
+
+        if all_errors:
+            print("VALIDATION FAILED:")
+            for wizard, errors in all_errors.items():
+                print(f"\n  {wizard}:")
+                for err in errors:
+                    print(f"    - {err}")
+            sys.exit(1)
+        else:
+            print("All wizard configs passed validation.")
+            sys.exit(0)
+    else:
+        config_path = sys.argv[1]
+        errors = validate_config(config_path)
+
+        if errors:
+            print(f"VALIDATION FAILED for {config_path}:")
+            for err in errors:
+                print(f"  - {err}")
+            sys.exit(1)
+        else:
+            print(f"PASSED: {config_path}")
+            sys.exit(0)
+
+
+if __name__ == "__main__":
+    main()

bin/agent-loop.sh

@@ -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

bin/claude-loop.sh

@@ -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
 

bin/deadman-fallback.py

@@ -0,0 +1,264 @@
+     1|#!/usr/bin/env python3
+     2|"""
+     3|Dead Man Switch Fallback Engine
+     4|
+     5|When the dead man switch triggers (zero commits for 2+ hours, model down,
+     6|Gitea unreachable, etc.), this script diagnoses the failure and applies
+     7|common sense fallbacks automatically.
+     8|
+     9|Fallback chain:
+    10|1. Primary model (Anthropic) down -> switch config to local-llama.cpp
+    11|2. Gitea unreachable -> cache issues locally, retry on recovery
+    12|3. VPS agents down -> alert + lazarus protocol
+    13|4. Local llama.cpp down -> try Ollama, then alert-only mode
+    14|5. All inference dead -> safe mode (cron pauses, alert Alexander)
+    15|
+    16|Each fallback is reversible. Recovery auto-restores the previous config.
+    17|"""
+    18|import os
+    19|import sys
+    20|import json
+    21|import subprocess
+    22|import time
+    23|import yaml
+    24|import shutil
+    25|from pathlib import Path
+    26|from datetime import datetime, timedelta
+    27|
+    28|HERMES_HOME = Path(os.environ.get("HERMES_HOME", os.path.expanduser("~/.hermes")))
+    29|CONFIG_PATH = HERMES_HOME / "config.yaml"
+    30|FALLBACK_STATE = HERMES_HOME / "deadman-fallback-state.json"
+    31|BACKUP_CONFIG = HERMES_HOME / "config.yaml.pre-fallback"
+    32|FORGE_URL = "https://forge.alexanderwhitestone.com"
+    33|
+    34|def load_config():
+    35|    with open(CONFIG_PATH) as f:
+    36|        return yaml.safe_load(f)
+    37|
+    38|def save_config(cfg):
+    39|    with open(CONFIG_PATH, "w") as f:
+    40|        yaml.dump(cfg, f, default_flow_style=False)
+    41|
+    42|def load_state():
+    43|    if FALLBACK_STATE.exists():
+    44|        with open(FALLBACK_STATE) as f:
+    45|            return json.load(f)
+    46|    return {"active_fallbacks": [], "last_check": None, "recovery_pending": False}
+    47|
+    48|def save_state(state):
+    49|    state["last_check"] = datetime.now().isoformat()
+    50|    with open(FALLBACK_STATE, "w") as f:
+    51|        json.dump(state, f, indent=2)
+    52|
+    53|def run(cmd, timeout=10):
+    54|    try:
+    55|        r = subprocess.run(cmd, shell=True, capture_output=True, text=True, timeout=timeout)
+    56|        return r.returncode, r.stdout.strip(), r.stderr.strip()
+    57|    except subprocess.TimeoutExpired:
+    58|        return -1, "", "timeout"
+    59|    except Exception as e:
+    60|        return -1, "", str(e)
+    61|
+    62|# ─── HEALTH CHECKS ───
+    63|
+    64|def check_anthropic():
+    65|    """Can we reach Anthropic API?"""
+    66|    key = os.environ.get("ANTHROPIC_API_KEY", "")
+    67|    if not key:
+    68|        # Check multiple .env locations
+    69|        for env_path in [HERMES_HOME / ".env", Path.home() / ".hermes" / ".env"]:
+    70|            if env_path.exists():
+    71|                for line in open(env_path):
+    72|                    line = line.strip()
+    73|                    if line.startswith("ANTHROPIC_API_KEY=***
+    74|                        key = line.split("=", 1)[1].strip().strip('"').strip("'")
+    75|                        break
+    76|            if key:
+    77|                break
+    78|    if not key:
+    79|        return False, "no API key"
+    80|    code, out, err = run(
+    81|        f'curl -s -o /dev/null -w "%{{http_code}}" -H "x-api-key: {key}" '
+    82|        f'-H "anthropic-version: 2023-06-01" '
+    83|        f'https://api.anthropic.com/v1/messages -X POST '
+    84|        f'-H "content-type: application/json" '
+    85|        f'-d \'{{"model":"claude-haiku-4-5-20251001","max_tokens":1,"messages":[{{"role":"user","content":"ping"}}]}}\' ',
+    86|        timeout=15
+    87|    )
+    88|    if code == 0 and out in ("200", "429"):
+    89|        return True, f"HTTP {out}"
+    90|    return False, f"HTTP {out} err={err[:80]}"
+    91|
+    92|def check_local_llama():
+    93|    """Is local llama.cpp serving?"""
+    94|    code, out, err = run("curl -s http://localhost:8081/v1/models", timeout=5)
+    95|    if code == 0 and "hermes" in out.lower():
+    96|        return True, "serving"
+    97|    return False, f"exit={code}"
+    98|
+    99|def check_ollama():
+   100|    """Is Ollama running?"""
+   101|    code, out, err = run("curl -s http://localhost:11434/api/tags", timeout=5)
+   102|    if code == 0 and "models" in out:
+   103|        return True, "running"
+   104|    return False, f"exit={code}"
+   105|
+   106|def check_gitea():
+   107|    """Can we reach the Forge?"""
+   108|    token_path = Path.home() / ".config" / "gitea" / "timmy-token"
+   109|    if not token_path.exists():
+   110|        return False, "no token"
+   111|    token = token_path.read_text().strip()
+   112|    code, out, err = run(
+   113|        f'curl -s -o /dev/null -w "%{{http_code}}" -H "Authorization: token {token}" '
+   114|        f'"{FORGE_URL}/api/v1/user"',
+   115|        timeout=10
+   116|    )
+   117|    if code == 0 and out == "200":
+   118|        return True, "reachable"
+   119|    return False, f"HTTP {out}"
+   120|
+   121|def check_vps(ip, name):
+   122|    """Can we SSH into a VPS?"""
+   123|    code, out, err = run(f"ssh -o ConnectTimeout=5 root@{ip} 'echo alive'", timeout=10)
+   124|    if code == 0 and "alive" in out:
+   125|        return True, "alive"
+   126|    return False, f"unreachable"
+   127|
+   128|# ─── FALLBACK ACTIONS ───
+   129|
+   130|def fallback_to_local_model(cfg):
+   131|    """Switch primary model from Anthropic to local llama.cpp"""
+   132|    if not BACKUP_CONFIG.exists():
+   133|        shutil.copy2(CONFIG_PATH, BACKUP_CONFIG)
+   134|    
+   135|    cfg["model"]["provider"] = "local-llama.cpp"
+   136|    cfg["model"]["default"] = "hermes3"
+   137|    save_config(cfg)
+   138|    return "Switched primary model to local-llama.cpp/hermes3"
+   139|
+   140|def fallback_to_ollama(cfg):
+   141|    """Switch to Ollama if llama.cpp is also down"""
+   142|    if not BACKUP_CONFIG.exists():
+   143|        shutil.copy2(CONFIG_PATH, BACKUP_CONFIG)
+   144|    
+   145|    cfg["model"]["provider"] = "ollama"
+   146|    cfg["model"]["default"] = "gemma4:latest"
+   147|    save_config(cfg)
+   148|    return "Switched primary model to ollama/gemma4:latest"
+   149|
+   150|def enter_safe_mode(state):
+   151|    """Pause all non-essential cron jobs, alert Alexander"""
+   152|    state["safe_mode"] = True
+   153|    state["safe_mode_entered"] = datetime.now().isoformat()
+   154|    save_state(state)
+   155|    return "SAFE MODE: All inference down. Cron jobs should be paused. Alert Alexander."
+   156|
+   157|def restore_config():
+   158|    """Restore pre-fallback config when primary recovers"""
+   159|    if BACKUP_CONFIG.exists():
+   160|        shutil.copy2(BACKUP_CONFIG, CONFIG_PATH)
+   161|        BACKUP_CONFIG.unlink()
+   162|        return "Restored original config from backup"
+   163|    return "No backup config to restore"
+   164|
+   165|# ─── MAIN DIAGNOSIS AND FALLBACK ENGINE ───
+   166|
+   167|def diagnose_and_fallback():
+   168|    state = load_state()
+   169|    cfg = load_config()
+   170|    
+   171|    results = {
+   172|        "timestamp": datetime.now().isoformat(),
+   173|        "checks": {},
+   174|        "actions": [],
+   175|        "status": "healthy"
+   176|    }
+   177|    
+   178|    # Check all systems
+   179|    anthropic_ok, anthropic_msg = check_anthropic()
+   180|    results["checks"]["anthropic"] = {"ok": anthropic_ok, "msg": anthropic_msg}
+   181|    
+   182|    llama_ok, llama_msg = check_local_llama()
+   183|    results["checks"]["local_llama"] = {"ok": llama_ok, "msg": llama_msg}
+   184|    
+   185|    ollama_ok, ollama_msg = check_ollama()
+   186|    results["checks"]["ollama"] = {"ok": ollama_ok, "msg": ollama_msg}
+   187|    
+   188|    gitea_ok, gitea_msg = check_gitea()
+   189|    results["checks"]["gitea"] = {"ok": gitea_ok, "msg": gitea_msg}
+   190|    
+   191|    # VPS checks
+   192|    vpses = [
+   193|        ("167.99.126.228", "Allegro"),
+   194|        ("143.198.27.163", "Ezra"),
+   195|        ("159.203.146.185", "Bezalel"),
+   196|    ]
+   197|    for ip, name in vpses:
+   198|        vps_ok, vps_msg = check_vps(ip, name)
+   199|        results["checks"][f"vps_{name.lower()}"] = {"ok": vps_ok, "msg": vps_msg}
+   200|    
+   201|    current_provider = cfg.get("model", {}).get("provider", "anthropic")
+   202|    
+   203|    # ─── FALLBACK LOGIC ───
+   204|    
+   205|    # Case 1: Primary (Anthropic) down, local available
+   206|    if not anthropic_ok and current_provider == "anthropic":
+   207|        if llama_ok:
+   208|            msg = fallback_to_local_model(cfg)
+   209|            results["actions"].append(msg)
+   210|            state["active_fallbacks"].append("anthropic->local-llama")
+   211|            results["status"] = "degraded_local"
+   212|        elif ollama_ok:
+   213|            msg = fallback_to_ollama(cfg)
+   214|            results["actions"].append(msg)
+   215|            state["active_fallbacks"].append("anthropic->ollama")
+   216|            results["status"] = "degraded_ollama"
+   217|        else:
+   218|            msg = enter_safe_mode(state)
+   219|            results["actions"].append(msg)
+   220|            results["status"] = "safe_mode"
+   221|    
+   222|    # Case 2: Already on fallback, check if primary recovered
+   223|    elif anthropic_ok and "anthropic->local-llama" in state.get("active_fallbacks", []):
+   224|        msg = restore_config()
+   225|        results["actions"].append(msg)
+   226|        state["active_fallbacks"].remove("anthropic->local-llama")
+   227|        results["status"] = "recovered"
+   228|    elif anthropic_ok and "anthropic->ollama" in state.get("active_fallbacks", []):
+   229|        msg = restore_config()
+   230|        results["actions"].append(msg)
+   231|        state["active_fallbacks"].remove("anthropic->ollama")
+   232|        results["status"] = "recovered"
+   233|    
+   234|    # Case 3: Gitea down — just flag it, work locally
+   235|    if not gitea_ok:
+   236|        results["actions"].append("WARN: Gitea unreachable — work cached locally until recovery")
+   237|        if "gitea_down" not in state.get("active_fallbacks", []):
+   238|            state["active_fallbacks"].append("gitea_down")
+   239|        results["status"] = max(results["status"], "degraded_gitea", key=lambda x: ["healthy", "recovered", "degraded_gitea", "degraded_local", "degraded_ollama", "safe_mode"].index(x) if x in ["healthy", "recovered", "degraded_gitea", "degraded_local", "degraded_ollama", "safe_mode"] else 0)
+   240|    elif "gitea_down" in state.get("active_fallbacks", []):
+   241|        state["active_fallbacks"].remove("gitea_down")
+   242|        results["actions"].append("Gitea recovered — resume normal operations")
+   243|    
+   244|    # Case 4: VPS agents down
+   245|    for ip, name in vpses:
+   246|        key = f"vps_{name.lower()}"
+   247|        if not results["checks"][key]["ok"]:
+   248|            results["actions"].append(f"ALERT: {name} VPS ({ip}) unreachable — lazarus protocol needed")
+   249|    
+   250|    save_state(state)
+   251|    return results
+   252|
+   253|if __name__ == "__main__":
+   254|    results = diagnose_and_fallback()
+   255|    print(json.dumps(results, indent=2))
+   256|    
+   257|    # Exit codes for cron integration
+   258|    if results["status"] == "safe_mode":
+   259|        sys.exit(2)
+   260|    elif results["status"].startswith("degraded"):
+   261|        sys.exit(1)
+   262|    else:
+   263|        sys.exit(0)
+   264|

bin/gemini-loop.sh

@@ -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
 

bin/genchi-genbutsu.sh

@@ -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

bin/kaizen-retro.sh

@@ -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" "$@"

bin/pr-checklist.py

@@ -0,0 +1,191 @@
+#!/usr/bin/env python3
+"""pr-checklist.py -- Automated PR quality gate for Gitea CI.
+
+Enforces the review standards that agents skip when left to self-approve.
+Runs in CI on every pull_request event. Exits non-zero on any failure.
+
+Checks:
+  1. PR has >0 file changes (no empty PRs)
+  2. PR branch is not behind base branch
+  3. PR does not bundle >3 unrelated issues
+  4. Changed .py files pass syntax check (python -c import)
+  5. Changed .sh files are executable
+  6. PR body references an issue number
+  7. At least 1 non-author review exists (warning only)
+
+Refs: #393 (PERPLEXITY-08), Epic #385
+"""
+from __future__ import annotations
+
+import json
+import os
+import re
+import subprocess
+import sys
+from pathlib import Path
+
+
+def fail(msg: str) -> None:
+    print(f"FAIL: {msg}", file=sys.stderr)
+
+
+def warn(msg: str) -> None:
+    print(f"WARN: {msg}", file=sys.stderr)
+
+
+def ok(msg: str) -> None:
+    print(f"  OK: {msg}")
+
+
+def get_changed_files() -> list[str]:
+    """Return list of files changed in this PR vs base branch."""
+    base = os.environ.get("GITHUB_BASE_REF", "main")
+    try:
+        result = subprocess.run(
+            ["git", "diff", "--name-only", f"origin/{base}...HEAD"],
+            capture_output=True, text=True, check=True,
+        )
+        return [f for f in result.stdout.strip().splitlines() if f]
+    except subprocess.CalledProcessError:
+        # Fallback: diff against HEAD~1
+        result = subprocess.run(
+            ["git", "diff", "--name-only", "HEAD~1"],
+            capture_output=True, text=True, check=True,
+        )
+        return [f for f in result.stdout.strip().splitlines() if f]
+
+
+def check_has_changes(files: list[str]) -> bool:
+    """Check 1: PR has >0 file changes."""
+    if not files:
+        fail("PR has 0 file changes. Empty PRs are not allowed.")
+        return False
+    ok(f"PR changes {len(files)} file(s)")
+    return True
+
+
+def check_not_behind_base() -> bool:
+    """Check 2: PR branch is not behind base."""
+    base = os.environ.get("GITHUB_BASE_REF", "main")
+    try:
+        result = subprocess.run(
+            ["git", "rev-list", "--count", f"HEAD..origin/{base}"],
+            capture_output=True, text=True, check=True,
+        )
+        behind = int(result.stdout.strip())
+        if behind > 0:
+            fail(f"Branch is {behind} commit(s) behind {base}. Rebase or merge.")
+            return False
+        ok(f"Branch is up-to-date with {base}")
+        return True
+    except (subprocess.CalledProcessError, ValueError):
+        warn("Could not determine if branch is behind base (git fetch may be needed)")
+        return True  # Don't block on CI fetch issues
+
+
+def check_issue_bundling(pr_body: str) -> bool:
+    """Check 3: PR does not bundle >3 unrelated issues."""
+    issue_refs = set(re.findall(r"#(\d+)", pr_body))
+    if len(issue_refs) > 3:
+        fail(f"PR references {len(issue_refs)} issues ({', '.join(sorted(issue_refs))}). "
+             "Max 3 per PR to prevent bundling. Split into separate PRs.")
+        return False
+    ok(f"PR references {len(issue_refs)} issue(s) (max 3)")
+    return True
+
+
+def check_python_syntax(files: list[str]) -> bool:
+    """Check 4: Changed .py files have valid syntax."""
+    py_files = [f for f in files if f.endswith(".py") and Path(f).exists()]
+    if not py_files:
+        ok("No Python files changed")
+        return True
+
+    all_ok = True
+    for f in py_files:
+        result = subprocess.run(
+            [sys.executable, "-c", f"import ast; ast.parse(open('{f}').read())"],
+            capture_output=True, text=True,
+        )
+        if result.returncode != 0:
+            fail(f"Syntax error in {f}: {result.stderr.strip()[:200]}")
+            all_ok = False
+
+    if all_ok:
+        ok(f"All {len(py_files)} Python file(s) pass syntax check")
+    return all_ok
+
+
+def check_shell_executable(files: list[str]) -> bool:
+    """Check 5: Changed .sh files are executable."""
+    sh_files = [f for f in files if f.endswith(".sh") and Path(f).exists()]
+    if not sh_files:
+        ok("No shell scripts changed")
+        return True
+
+    all_ok = True
+    for f in sh_files:
+        if not os.access(f, os.X_OK):
+            fail(f"{f} is not executable. Run: chmod +x {f}")
+            all_ok = False
+
+    if all_ok:
+        ok(f"All {len(sh_files)} shell script(s) are executable")
+    return all_ok
+
+
+def check_issue_reference(pr_body: str) -> bool:
+    """Check 6: PR body references an issue number."""
+    if re.search(r"#\d+", pr_body):
+        ok("PR body references at least one issue")
+        return True
+    fail("PR body does not reference any issue (e.g. #123). "
+         "Every PR must trace to an issue.")
+    return False
+
+
+def main() -> int:
+    print("=" * 60)
+    print("PR Checklist — Automated Quality Gate")
+    print("=" * 60)
+    print()
+
+    # Get PR body from env or git log
+    pr_body = os.environ.get("PR_BODY", "")
+    if not pr_body:
+        try:
+            result = subprocess.run(
+                ["git", "log", "--format=%B", "-1"],
+                capture_output=True, text=True, check=True,
+            )
+            pr_body = result.stdout
+        except subprocess.CalledProcessError:
+            pr_body = ""
+
+    files = get_changed_files()
+    failures = 0
+
+    checks = [
+        check_has_changes(files),
+        check_not_behind_base(),
+        check_issue_bundling(pr_body),
+        check_python_syntax(files),
+        check_shell_executable(files),
+        check_issue_reference(pr_body),
+    ]
+
+    failures = sum(1 for c in checks if not c)
+
+    print()
+    print("=" * 60)
+    if failures:
+        print(f"RESULT: {failures} check(s) FAILED")
+        print("Fix the issues above and push again.")
+        return 1
+    else:
+        print("RESULT: All checks passed")
+        return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

cron/jobs-backup-2026-04-10.json

@@ -0,0 +1,212 @@
+[
+  {
+    "job_id": "9e0624269ba7",
+    "name": "Triage Heartbeat",
+    "schedule": "every 15m",
+    "state": "paused"
+  },
+  {
+    "job_id": "e29eda4a8548",
+    "name": "PR Review Sweep",
+    "schedule": "every 30m",
+    "state": "scheduled"
+  },
+  {
+    "job_id": "a77a87392582",
+    "name": "Health Monitor",
+    "schedule": "every 5m",
+    "state": "scheduled"
+  },
+  {
+    "job_id": "5e9d952871bc",
+    "name": "Agent Status Check",
+    "schedule": "every 10m",
+    "state": "paused"
+  },
+  {
+    "job_id": "36fb2f630a17",
+    "name": "Hermes Philosophy Loop",
+    "schedule": "every 1440m",
+    "state": "paused"
+  },
+  {
+    "job_id": "b40a96a2f48c",
+    "name": "wolf-eval-cycle",
+    "schedule": "every 240m",
+    "state": "paused"
+  },
+  {
+    "job_id": "4204e568b862",
+    "name": "Burn Mode \u2014 Timmy Orchestrator",
+    "schedule": "every 15m",
+    "state": "scheduled"
+  },
+  {
+    "job_id": "0944a976d034",
+    "name": "Burn Mode",
+    "schedule": "every 15m",
+    "state": "paused"
+  },
+  {
+    "job_id": "62016b960fa0",
+    "name": "velocity-engine",
+    "schedule": "every 30m",
+    "state": "paused"
+  },
+  {
+    "job_id": "e9d49eeff79c",
+    "name": "weekly-skill-extraction",
+    "schedule": "every 10080m",
+    "state": "scheduled"
+  },
+  {
+    "job_id": "75c74a5bb563",
+    "name": "tower-tick",
+    "schedule": "every 1m",
+    "state": "scheduled"
+  },
+  {
+    "job_id": "390a19054d4c",
+    "name": "Burn Deadman",
+    "schedule": "every 30m",
+    "state": "scheduled"
+  },
+  {
+    "job_id": "05e3c13498fa",
+    "name": "Morning Report \u2014 Burn Mode",
+    "schedule": "0 6 * * *",
+    "state": "scheduled"
+  },
+  {
+    "job_id": "64fe44b512b9",
+    "name": "evennia-morning-report",
+    "schedule": "0 9 * * *",
+    "state": "scheduled"
+  },
+  {
+    "job_id": "3896a7fd9747",
+    "name": "Gitea Priority Inbox",
+    "schedule": "every 3m",
+    "state": "scheduled"
+  },
+  {
+    "job_id": "f64c2709270a",
+    "name": "Config Drift Guard",
+    "schedule": "every 30m",
+    "state": "scheduled"
+  },
+  {
+    "job_id": "fc6a75b7102a",
+    "name": "Gitea Event Watcher",
+    "schedule": "every 2m",
+    "state": "scheduled"
+  },
+  {
+    "job_id": "12e59648fb06",
+    "name": "Burndown Night Watcher",
+    "schedule": "every 15m",
+    "state": "scheduled"
+  },
+  {
+    "job_id": "35d3ada9cf8f",
+    "name": "Mempalace Forge \u2014 Issue Analysis",
+    "schedule": "every 60m",
+    "state": "scheduled"
+  },
+  {
+    "job_id": "190b6fb8dc91",
+    "name": "Mempalace Watchtower \u2014 Fleet Health",
+    "schedule": "every 30m",
+    "state": "scheduled"
+  },
+  {
+    "job_id": "710ab589813c",
+    "name": "Ezra Health Monitor",
+    "schedule": "every 15m",
+    "state": "scheduled"
+  },
+  {
+    "job_id": "a0a9cce4575c",
+    "name": "daily-poka-yoke-ultraplan-awesometools",
+    "schedule": "every 1440m",
+    "state": "scheduled"
+  },
+  {
+    "job_id": "adc3a51457bd",
+    "name": "vps-agent-dispatch",
+    "schedule": "every 10m",
+    "state": "scheduled"
+  },
+  {
+    "job_id": "afd2c4eac44d",
+    "name": "Project Mnemosyne Nightly Burn v2",
+    "schedule": "*/30 * * * *",
+    "state": "scheduled"
+  },
+  {
+    "job_id": "f3a3c2832af0",
+    "name": "gemma4-multimodal-worker",
+    "schedule": "once in 15m",
+    "state": "completed"
+  },
+  {
+    "job_id": "c17a85c19838",
+    "name": "know-thy-father-analyzer",
+    "schedule": "0 * * * *",
+    "state": "scheduled"
+  },
+  {
+    "job_id": "2490fc01a14d",
+    "name": "Testament Burn - 10min work loop",
+    "schedule": "*/10 * * * *",
+    "state": "scheduled"
+  },
+  {
+    "job_id": "f5e858159d97",
+    "name": "Timmy Foundation Burn \u2014 15min PR loop",
+    "schedule": "*/15 * * * *",
+    "state": "scheduled"
+  },
+  {
+    "job_id": "5e262fb9bdce",
+    "name": "nightwatch-health-monitor",
+    "schedule": "*/15 * * * *",
+    "state": "scheduled"
+  },
+  {
+    "job_id": "f2b33a9dcf96",
+    "name": "nightwatch-mempalace-mine",
+    "schedule": "0 */2 * * *",
+    "state": "scheduled"
+  },
+  {
+    "job_id": "82cb9e76c54d",
+    "name": "nightwatch-backlog-burn",
+    "schedule": "0 */4 * * *",
+    "state": "scheduled"
+  },
+  {
+    "job_id": "d20e42a52863",
+    "name": "beacon-sprint",
+    "schedule": "*/15 * * * *",
+    "state": "scheduled"
+  },
+  {
+    "job_id": "579269489961",
+    "name": "testament-story",
+    "schedule": "*/15 * * * *",
+    "state": "scheduled"
+  },
+  {
+    "job_id": "2e5f9140d1ab",
+    "name": "nightwatch-research",
+    "schedule": "0 */2 * * *",
+    "state": "scheduled"
+  },
+  {
+    "job_id": "aeba92fd65e6",
+    "name": "timmy-dreams",
+    "schedule": "30 5 * * *",
+    "state": "scheduled"
+  }
+]

cron/jobs.json

@@ -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"
-}
+}

cron/vps/allegro-crontab-backup.txt

@@ -0,0 +1,14 @@
+0 6 * * * /bin/bash /root/wizards/scripts/model_download_guard.sh >> /var/log/model_guard.log 2>&1
+
+# Allegro Hybrid Heartbeat — quick wins every 15 min
+*/15 * * * * /usr/bin/python3 /root/allegro/heartbeat_daemon.py >> /var/log/allegro_heartbeat.log 2>&1
+
+# Allegro Burn Mode Cron Jobs - Deployed via issue #894
+
+0 6 * * * cd /root/.hermes && python3 -c "import hermes_agent; from hermes_tools import terminal; output = terminal('echo \"Morning Report: $(date)\"'); print(output.get('output', ''))" >> /root/.hermes/logs/morning-report-$(date +\%Y\%m\%d).log 2>&1 # Allegro Morning Report at 0600
+
+0,30 * * * * cd /root/.hermes && python3 /root/.hermes/retry_wrapper.py "python3 allegro/quick-lane-check.py" >> burn-logs/quick-lane-$(date +\%Y\%m\%d).log 2>&1 # Allegro Burn Loop #1 (with retry)
+15,45 * * * * cd /root/.hermes && python3 /root/.hermes/retry_wrapper.py "python3 allegro/burn-mode-validator.py" >> burn-logs/validator-$(date +\%Y\%m\%d).log 2>&1 # Allegro Burn Loop #2 (with retry)
+
+*/2 * * * * /root/wizards/bezalel/dead_man_monitor.sh
+*/2 * * * * /root/wizards/allegro/bin/config-deadman.sh

cron/vps/bezalel-crontab-backup.txt

@@ -0,0 +1,10 @@
+0 2 * * * /root/wizards/bezalel/run_nightly_watch.sh
+0 3 * * * /root/wizards/bezalel/mempalace_nightly.sh
+*/10 * * * * pgrep -f "act_runner daemon" > /dev/null || (cd /opt/gitea-runner && nohup ./act_runner daemon > /var/log/gitea-runner.log 2>&1 &)
+30 3 * * * /root/wizards/bezalel/backup_databases.sh
+*/15 * * * * /root/wizards/bezalel/meta_heartbeat.sh
+0 4 * * * /root/wizards/bezalel/secret_guard.sh
+0 4 * * * /usr/bin/env bash /root/timmy-home/scripts/backup_pipeline.sh >> /var/log/timmy/backup_pipeline_cron.log 2>&1
+0 6 * * * /usr/bin/python3 /root/wizards/bezalel/ultraplan.py >> /var/log/bezalel-ultraplan.log 2>&1
+@reboot /root/wizards/bezalel/emacs-daemon-start.sh
+@reboot /root/wizards/bezalel/ngircd-start.sh

cron/vps/ezra-crontab-backup.txt

@@ -0,0 +1,13 @@
+# Burn Mode Cycles — 15 min autonomous loops
+*/15 * * * * /root/wizards/ezra/bin/burn-mode.sh >> /root/wizards/ezra/reports/burn-cron.log 2>&1
+
+# Household Snapshots — automated heartbeats and snapshots
+# Ezra Self-Improvement Automation Suite
+*/5 * * * * /usr/bin/python3 /root/wizards/ezra/tools/gitea_monitor.py >> /root/wizards/ezra/reports/gitea-monitor.log 2>&1
+*/5 * * * * /usr/bin/python3 /root/wizards/ezra/tools/awareness_loop.py >> /root/wizards/ezra/reports/awareness-loop.log 2>&1
+*/10 * * * * /usr/bin/python3 /root/wizards/ezra/tools/cron_health_monitor.py >> /root/wizards/ezra/reports/cron-health.log 2>&1
+0 6 * * * /usr/bin/python3 /root/wizards/ezra/tools/morning_kt_compiler.py >> /root/wizards/ezra/reports/morning-kt.log 2>&1
+5 6 * * * /usr/bin/python3 /root/wizards/ezra/tools/burndown_generator.py >> /root/wizards/ezra/reports/burndown.log 2>&1
+0 3 * * * /root/wizards/ezra/mempalace_nightly.sh >> /var/log/ezra_mempalace_cron.log 2>&1
+*/15 * * * * GITEA_TOKEN=6de6aa...1117 /root/wizards/ezra/dispatch-direct.sh >> /root/wizards/ezra/dispatch-cron.log 2>&1
+

docs/FLEET_BEHAVIOUR_HARDENING.md

@@ -0,0 +1,110 @@
+# Fleet Behaviour Hardening — Review & Action Plan
+
+**Author:** @perplexity  
+**Date:** 2026-04-08  
+**Context:** Alexander asked: "Is it the memory system or the behaviour guardrails?"  
+**Answer:** It's the guardrails. The memory system is adequate. The enforcement machinery is aspirational.
+
+---
+
+## Diagnosis: Why the Fleet Isn't Smart Enough
+
+After auditing SOUL.md, config.yaml, all 8 playbooks, the orchestrator, the guard scripts, and the v7.0.0 checkin, the pattern is clear:
+
+**The fleet has excellent design documents and broken enforcement.**
+
+| Layer | Design Quality | Enforcement Quality | Gap |
+|---|---|---|---|
+| SOUL.md | Excellent | None — no code reads it at runtime | Philosophy without machinery |
+| Playbooks (7 yaml) | Good lane map | Not invoked by orchestrator | Playbooks exist but nobody calls them |
+| Guard scripts (9) | Solid code | 1 of 9 wired (#395 audit) | 89% of guards are dead code |
+| Orchestrator | Sound design | Gateway dispatch is a no-op (#391) | Assigns issues but doesn't trigger work |
+| Cycle Guard | Good 10-min rule | No cron/loop calls it | Discipline without enforcement |
+| PR Reviewer | Clear rules | Runs every 30m (if scheduled) | Only guard that might actually fire |
+| Memory (MemPalace) | Working code | Retrieval enforcer wired | Actually operational |
+
+### The Core Problem
+
+Agents pick up issues and produce output, but there is **no pre-task checklist** and **no post-task quality gate**. An agent can:
+
+1. Start work without checking if someone else already did it
+2. Produce output without running tests
+3. Submit a PR without verifying it addresses the issue
+4. Work for hours on something out of scope
+5. Create duplicate branches/PRs without detection
+
+The SOUL.md says "grounding before generation" but no code enforces it.  
+The playbooks define lanes but the orchestrator doesn't load them.  
+The guards exist but nothing calls them.
+
+---
+
+## What the Fleet Needs (Priority Order)
+
+### 1. Pre-Task Gate (MISSING — this PR adds it)
+
+Before an agent starts any issue:
+- [ ] Check if issue is already assigned to another agent
+- [ ] Check if a branch already exists for this issue
+- [ ] Check if a PR already exists for this issue  
+- [ ] Load relevant MemPalace context (retrieval enforcer)
+- [ ] Verify the agent has the right lane for this work (playbook check)
+
+### 2. Post-Task Gate (MISSING — this PR adds it)
+
+Before an agent submits a PR:
+- [ ] Verify the diff addresses the issue title/body
+- [ ] Run syntax_guard.py on changed files
+- [ ] Check for duplicate PRs targeting the same issue
+- [ ] Verify branch name follows convention
+- [ ] Run tests if they exist for changed files
+
+### 3. Wire the Existing Guards (8 of 9 are dead code)
+
+Per #395 audit:
+- Pre-commit hooks: need symlink on every machine
+- Cycle guard: need cron/loop integration  
+- Forge health check: need cron entry
+- Smoke test + deploy validate: need deploy script integration
+
+### 4. Orchestrator Dispatch Actually Works
+
+Per #391 audit: the orchestrator scores and assigns but the gateway dispatch just writes to `/tmp/hermes-dispatch.log`. Nobody reads that file. The dispatch needs to either:
+- Trigger `hermes` CLI on the target machine, or
+- Post a webhook that the agent loop picks up
+
+### 5. Agent Self-Assessment Loop
+
+After completing work, agents should answer:
+- Did I address the issue as stated?
+- Did I stay in scope?
+- Did I check the palace for prior work?
+- Did I run verification?
+
+This is what SOUL.md calls "the apparatus that gives these words teeth."
+
+---
+
+## What's Working (Don't Touch)
+
+- **MemPalace sovereign_store.py** — SQLite + FTS5 + HRR, operational
+- **Retrieval enforcer** — wired to SovereignStore as of 14 hours ago
+- **Wake-up protocol** — palace-first boot sequence
+- **PR reviewer playbook** — clear rules, well-scoped
+- **Issue triager playbook** — comprehensive lane map with 11 agents
+- **Cycle guard code** — solid 10-min slice discipline (just needs wiring)
+- **Config drift guard** — active cron, working
+- **Dead man switch** — active, working
+
+---
+
+## Recommendation
+
+The memory system is not the bottleneck. The behaviour guardrails are. Specifically:
+
+1. **Add `task_gate.py`** — pre-task and post-task quality gates that every agent loop calls
+2. **Wire cycle_guard.py** — add start/complete calls to agent loop
+3. **Wire pre-commit hooks** — deploy script should symlink on provision
+4. **Fix orchestrator dispatch** — make it actually trigger work, not just log
+
+This PR adds item 1. Items 2-4 need SSH access and are flagged for Timmy/Allegro.

docs/MEMORY_ARCHITECTURE.md

@@ -0,0 +1,141 @@
+# Memory Architecture
+
+> How Timmy remembers, recalls, and learns — without hallucinating.
+
+Refs: Epic #367 | Sub-issues #368, #369, #370, #371, #372
+
+## Overview
+
+Timmy's memory system uses a **Memory Palace** architecture — a structured, file-backed knowledge store organized into rooms and drawers. When faced with a recall question, the agent checks its palace *before* generating from scratch.
+
+This document defines the retrieval order, storage layers, and data flow that make this work.
+
+## Retrieval Order (L0–L5)
+
+When the agent receives a prompt that looks like a recall question ("what did we do?", "what's the status of X?"), the retrieval enforcer intercepts it and walks through layers in order:
+
+| Layer | Source | Question Answered | Short-circuits? |
+|-------|--------|-------------------|------------------|
+| L0 | `identity.txt` | Who am I? What are my mandates? | No (always loaded) |
+| L1 | Palace rooms/drawers | What do I know about this topic? | Yes, if hit |
+| L2 | Session scratchpad | What have I learned this session? | Yes, if hit |
+| L3 | Artifact retrieval (Gitea API) | Can I fetch the actual issue/file/log? | Yes, if hit |
+| L4 | Procedures/playbooks | Is there a documented way to do this? | Yes, if hit |
+| L5 | Free generation | (Only when L0–L4 are exhausted) | N/A |
+
+**Key principle:** The agent never reaches L5 (free generation) if any prior layer has relevant data. This eliminates hallucination for recall-style queries.
+
+## Storage Layout
+
+```
+~/.mempalace/
+  identity.txt              # L0: Who I am, mandates, personality
+  rooms/
+    projects/
+      timmy-config.md        # What I know about timmy-config
+      hermes-agent.md        # What I know about hermes-agent
+    people/
+      alexander.md           # Working relationship context
+    architecture/
+      fleet.md               # Fleet system knowledge
+      mempalace.md           # Self-knowledge about this system
+  config/
+    mempalace.yaml           # Palace configuration
+
+~/.hermes/
+  scratchpad/
+    {session_id}.json        # L2: Ephemeral session context
+```
+
+## Components
+
+### 1. Memory Palace Skill (`mempalace.py`) — #368
+
+Core data structures:
+- `PalaceRoom`: A named collection of drawers (topics)
+- `Mempalace`: The top-level palace with room management
+- Factory constructors: `for_issue_analysis()`, `for_health_check()`, `for_code_review()`
+
+### 2. Retrieval Enforcer (`retrieval_enforcer.py`) — #369
+
+Middleware that intercepts recall-style prompts:
+1. Detects recall patterns ("what did", "status of", "last time we")
+2. Walks L0→L4 in order, short-circuiting on first hit
+3. Only allows free generation (L5) when all layers return empty
+4. Produces an honest fallback: "I don't have this in my memory palace."
+
+### 3. Session Scratchpad (`scratchpad.py`) — #370
+
+Ephemeral, session-scoped working memory:
+- Write-append only during a session
+- Entries have TTL (default: 1 hour)
+- Queried at L2 in retrieval chain
+- Never auto-promoted to palace
+
+### 4. Memory Promotion — #371
+
+Explicit promotion from scratchpad to palace:
+- Agent must call `promote_to_palace()` with a reason
+- Dedup check against target drawer
+- Summary required (raw tool output never stored)
+- Conflict detection when new memory contradicts existing
+
+### 5. Wake-Up Protocol (`wakeup.py`) — #372
+
+Boot sequence for new sessions:
+```
+Session Start
+  │
+  ├─ L0: Load identity.txt
+  ├─ L1: Scan palace rooms for active context
+  ├─ L1.5: Surface promoted memories from last session
+  ├─ L2: Load surviving scratchpad entries
+  │
+  └─ Ready: agent knows who it is, what it was doing, what it learned
+```
+
+## Data Flow
+
+```
+              ┌──────────────────┐
+              │  User Prompt     │
+              └────────┬─────────┘
+                       │
+              ┌────────┴─────────┐
+              │ Recall Detector  │
+              └────┬───────┬─────┘
+                   │           │
+            [recall]     [not recall]
+                   │           │
+          ┌───────┴────┐    ┌──┬─┴───────┐
+          │ Retrieval  │    │ Normal Flow │
+          │ Enforcer   │    └─────────────┘
+          │ L0→L1→L2  │
+          │ →L3→L4→L5│
+          └──────┬─────┘
+                 │
+          ┌──────┴─────┐
+          │  Response    │
+          │ (grounded)  │
+          └────────────┘
+```
+
+## Anti-Patterns
+
+| Don't | Do Instead |
+|-------|------------|
+| Generate from vibes when palace has data | Check palace first (L1) |
+| Auto-promote everything to palace | Require explicit `promote_to_palace()` with reason |
+| Store raw API responses as memories | Summarize before storing |
+| Hallucinate when palace is empty | Say "I don't have this in my memory palace" |
+| Dump entire palace on wake-up | Selective loading based on session context |
+
+## Status
+
+| Component | Issue | PR | Status |
+|-----------|-------|----|--------|
+| Skill port | #368 | #374 | In Review |
+| Retrieval enforcer | #369 | #374 | In Review |
+| Session scratchpad | #370 | #374 | In Review |
+| Memory promotion | #371 | — | Open |
+| Wake-up protocol | #372 | #374 | In Review |

evaluations/crewai/.gitignore

@@ -0,0 +1,4 @@
+venv/
+__pycache__/
+*.pyc
+.env

evaluations/crewai/CREWAI_EVALUATION.md

@@ -0,0 +1,140 @@
+# CrewAI Evaluation for Phase 2 Integration
+
+**Date:** 2026-04-07  
+**Issue:** [#358 ORCHESTRATOR-4] Evaluate CrewAI for Phase 2 integration  
+**Author:** Ezra  
+**House:** hermes-ezra
+
+## Summary
+
+CrewAI was installed, a 2-agent proof-of-concept crew was built, and an operational test was attempted against issue #358. Based on code analysis, installation experience, and alignment with the coordinator-first protocol, the **verdict is REJECT for Phase 2 integration**. CrewAI adds significant dependency weight and abstraction opacity without solving problems the current Huey-based stack cannot already handle.
+
+---
+
+## 1. Proof-of-Concept Crew
+
+### Agents
+
+| Agent | Role | Responsibility |
+|-------|------|----------------|
+| `researcher` | Orchestration Researcher | Reads current orchestrator files and extracts factual comparisons |
+| `evaluator` | Integration Evaluator | Synthesizes research into a structured adoption recommendation |
+
+### Tools
+
+- `read_orchestrator_files` — Returns `orchestration.py`, `tasks.py`, `bin/timmy-orchestrator.sh`, and `docs/coordinator-first-protocol.md`
+- `read_issue_358` — Returns the text of the governing issue
+
+### Code
+
+See `poc_crew.py` in this directory for the full implementation.
+
+---
+
+## 2. Operational Test Results
+
+### What worked
+- `pip install crewai` completed successfully (v1.13.0)
+- Agent and tool definitions compiled without errors
+- Crew startup and task dispatch UI rendered correctly
+
+### What failed
+- **Live LLM execution blocked by authentication failures.** Available API credentials (OpenRouter, Kimi) were either rejected or not present in the runtime environment.
+- No local `llama-server` was running on the expected port (8081), and starting one was out of scope for this evaluation.
+
+### Why this matters
+The authentication failure is **not a trivial setup issue** — it is a preview of the operational complexity CrewAI introduces. The current Huey stack runs entirely offline against local SQLite and local Hermes models. CrewAI, by contrast, demands either:
+- A managed cloud LLM API with live credentials, or
+- A carefully tuned local model endpoint that supports its verbose ReAct-style prompts
+
+Either path increases blast radius and failure modes.
+
+---
+
+## 3. Current Custom Orchestrator Analysis
+
+### Stack
+- **Huey** (`orchestration.py`) — SQLite-backed task queue, ~6 lines of initialization
+- **tasks.py** — ~2,300 lines of scheduled work (triage, PR review, metrics, heartbeat)
+- **bin/timmy-orchestrator.sh** — Shell-based polling loop for state gathering and PR review
+- **docs/coordinator-first-protocol.md** — Intake → Triage → Route → Track → Verify → Report
+
+### Strengths
+1. **Sovereignty** — No external SaaS dependency for queue execution. SQLite is local and inspectable.
+2. **Gitea as truth** — All state mutations are visible in the forge. Local-only state is explicitly advisory.
+3. **Simplicity** — Huey has a tiny surface area. A human can read `orchestration.py` in seconds.
+4. **Tool-native** — `tasks.py` calls Hermes directly via `subprocess.run([HERMES_PYTHON, ...])`. No framework indirection.
+5. **Deterministic routing** — The coordinator-first protocol defines exact authority boundaries (Timmy, Allegro, workers, Alexander).
+
+### Gaps
+- **No built-in agent memory/RAG** — but this is intentional per the pre-compaction flush contract and memory-continuity doctrine.
+- **No multi-agent collaboration primitives** — but the current stack routes work to single owners explicitly.
+- **PR review is shell-prompt driven** — Could be tightened, but this is a prompt engineering issue, not an orchestrator gap.
+
+---
+
+## 4. CrewAI Capability Analysis
+
+### What CrewAI offers
+- **Agent roles** — Declarative backstory/goal/role definitions
+- **Task graphs** — Sequential, hierarchical, or parallel task execution
+- **Tool registry** — Pydantic-based tool schemas with auto-validation
+- **Memory/RAG** — Built-in short-term and long-term memory via ChromaDB/LanceDB
+- **Crew-wide context sharing** — Output from one task flows to the next
+
+### Dependency footprint observed
+CrewAI pulled in **85+ packages**, including:
+- `chromadb` (~20 MB) + `onnxruntime` (~17 MB)
+- `lancedb` (~47 MB)
+- `kubernetes` client (unused but required by Chroma)
+- `grpcio`, `opentelemetry-*`, `pdfplumber`, `textual`
+
+Total venv size: **>500 MB**.
+
+By contrast, Huey is **one package** (`huey`) with zero required services.
+
+---
+
+## 5. Alignment with Coordinator-First Protocol
+
+| Principle | Current Stack | CrewAI | Assessment |
+|-----------|--------------|--------|------------|
+| **Gitea is truth** | All assignments, PRs, comments are explicit API calls | Agent memory is local/ChromaDB. State can drift from Gitea unless every tool explicitly syncs | **Misaligned** |
+| **Local-only state is advisory** | SQLite queue is ephemeral; canonical state is in Gitea | CrewAI encourages "crew memory" as authoritative | **Misaligned** |
+| **Verification-before-complete** | PR review + merge require visible diffs and explicit curl calls | Tool outputs can be hallucinated or incomplete without strict guardrails | **Requires heavy customization** |
+| **Sovereignty** | Runs on VPS with no external orchestrator SaaS | Requires external LLM or complex local model tuning | **Degraded** |
+| **Simplicity** | ~6 lines for Huey init, readable shell scripts | 500+ MB dependency tree, opaque LangChain-style internals | **Degraded** |
+
+---
+
+## 6. Verdict
+
+**REJECT CrewAI for Phase 2 integration.**
+
+**Confidence:** High
+
+### Trade-offs
+- **Pros of CrewAI:** Nice agent-role syntax; built-in task sequencing; rich tool schema validation; active ecosystem.
+- **Cons of CrewAI:** Massive dependency footprint; memory model conflicts with Gitea-as-truth doctrine; requires either cloud API spend or fragile local model integration; adds abstraction layers that obscure what is actually happening.
+
+### Risks if adopted
+1. **Dependency rot** — 85+ transitive dependencies, many with conflicting version ranges.
+2. **State drift** — CrewAI's memory primitives train users to treat local vector DB as truth.
+3. **Credential fragility** — Live API requirements introduce a new failure mode the current stack does not have.
+4. **Vendor-like lock-in** — CrewAI's abstractions sit thickly over LangChain. Debugging a stuck crew is harder than debugging a Huey task traceback.
+
+### Recommended next step
+Instead of adopting CrewAI, **evolve the current Huey stack** with:
+1. A lightweight `Agent` dataclass in `tasks.py` (role, goal, system_prompt) to get the organizational clarity of CrewAI without the framework weight.
+2. A `delegate()` helper that uses Hermes's existing `delegate_tool.py` for multi-agent work.
+3. Keep Gitea as the only durable state surface. Any "memory" should flush to issue comments or `timmy-home` markdown, not a vector DB.
+
+If multi-agent collaboration becomes a hard requirement in the future, evaluate lighter alternatives (e.g., raw OpenAI/Anthropic function-calling loops, or a thin `smolagents`-style wrapper) before reconsidering CrewAI.
+
+---
+
+## Artifacts
+
+- `poc_crew.py` — 2-agent CrewAI proof-of-concept
+- `requirements.txt` — Dependency manifest
+- `CREWAI_EVALUATION.md` — This document

evaluations/crewai/poc_crew.py

@@ -0,0 +1,150 @@
+#!/usr/bin/env python3
+"""CrewAI proof-of-concept for evaluating Phase 2 orchestrator integration.
+
+Tests CrewAI against a real issue: #358 [ORCHESTRATOR-4] Evaluate CrewAI
+for Phase 2 integration.
+"""
+
+import os
+from pathlib import Path
+from crewai import Agent, Task, Crew, LLM
+from crewai.tools import BaseTool
+
+# ── Configuration ─────────────────────────────────────────────────────
+
+OPENROUTER_API_KEY = os.getenv(
+    "OPENROUTER_API_KEY",
+    "dsk-or-v1-f60c89db12040267458165cf192e815e339eb70548e4a0a461f5f0f69e6ef8b0",
+)
+
+llm = LLM(
+    model="openrouter/google/gemini-2.0-flash-001",
+    api_key=OPENROUTER_API_KEY,
+    base_url="https://openrouter.ai/api/v1",
+)
+
+REPO_ROOT = Path(__file__).resolve().parents[2]
+
+
+def _slurp(relpath: str, max_lines: int = 150) -> str:
+    p = REPO_ROOT / relpath
+    if not p.exists():
+        return f"[FILE NOT FOUND: {relpath}]"
+    lines = p.read_text().splitlines()
+    header = f"=== {relpath} ({len(lines)} lines total, showing first {max_lines}) ===\n"
+    return header + "\n".join(lines[:max_lines])
+
+
+# ── Tools ─────────────────────────────────────────────────────────────
+
+class ReadOrchestratorFilesTool(BaseTool):
+    name: str = "read_orchestrator_files"
+    description: str = (
+        "Reads the current custom orchestrator implementation files "
+        "(orchestration.py, tasks.py, timmy-orchestrator.sh, coordinator-first-protocol.md) "
+        "and returns their contents for analysis."
+    )
+
+    def _run(self) -> str:
+        return "\n\n".join(
+            [
+                _slurp("orchestration.py"),
+                _slurp("tasks.py", max_lines=120),
+                _slurp("bin/timmy-orchestrator.sh", max_lines=120),
+                _slurp("docs/coordinator-first-protocol.md", max_lines=120),
+            ]
+        )
+
+
+class ReadIssueTool(BaseTool):
+    name: str = "read_issue_358"
+    description: str = "Returns the text of Gitea issue #358 that we are evaluating."
+
+    def _run(self) -> str:
+        return (
+            "Title: [ORCHESTRATOR-4] Evaluate CrewAI for Phase 2 integration\n"
+            "Body:\n"
+            "Part of Epic: #354\n\n"
+            "Install CrewAI, build a proof-of-concept crew with 2 agents, "
+            "test on a real issue. Evaluate: does it add value over our custom orchestrator? Document findings."
+        )
+
+
+# ── Agents ────────────────────────────────────────────────────────────
+
+researcher = Agent(
+    role="Orchestration Researcher",
+    goal="Gather a complete understanding of the current custom orchestrator and how CrewAI compares to it.",
+    backstory=(
+        "You are a systems architect who specializes in evaluating orchestration frameworks. "
+        "You read code carefully, extract facts, and avoid speculation. "
+        "You focus on concrete capabilities, dependencies, and operational complexity."
+    ),
+    llm=llm,
+    tools=[ReadOrchestratorFilesTool(), ReadIssueTool()],
+    verbose=True,
+)
+
+evaluator = Agent(
+    role="Integration Evaluator",
+    goal="Synthesize research into a clear recommendation on whether CrewAI adds value for Phase 2.",
+    backstory=(
+        "You are a pragmatic engineering lead who values sovereignty, simplicity, and observable state. "
+        "You compare frameworks against the team's existing coordinator-first protocol. "
+        "You produce structured recommendations with explicit trade-offs."
+    ),
+    llm=llm,
+    verbose=True,
+)
+
+# ── Tasks ─────────────────────────────────────────────────────────────
+
+task_research = Task(
+    description=(
+        "Read the current custom orchestrator files and issue #358. "
+        "Produce a structured research report covering:\n"
+        "1. Current stack summary (Huey + tasks.py + timmy-orchestrator.sh)\n"
+        "2. Current strengths (sovereignty, local-first, Gitea as truth, simplicity)\n"
+        "3. Current gaps or limitations (if any)\n"
+        "4. What CrewAI offers (agent roles, tasks, crews, tools, memory/RAG)\n"
+        "5. CrewAI's dependencies and operational footprint (what you observed during installation)\n"
+        "Be factual and concise."
+    ),
+    expected_output="A structured markdown research report with the 5 sections above.",
+    agent=researcher,
+)
+
+task_evaluate = Task(
+    description=(
+        "Using the research report, evaluate whether CrewAI should be adopted for Phase 2 integration. "
+        "Consider the coordinator-first protocol (Gitea as truth, local-only state is advisory, "
+        "verification-before-complete, sovereignty).\n\n"
+        "Produce a final evaluation with:\n"
+        "- VERDICT: Adopt / Reject / Defer\n"
+        "- Confidence: High / Medium / Low\n"
+        "- Key trade-offs (3-5 bullets)\n"
+        "- Risks if adopted\n"
+        "- Recommended next step"
+    ),
+    expected_output="A structured markdown evaluation with verdict, confidence, trade-offs, risks, and recommendation.",
+    agent=evaluator,
+    context=[task_research],
+)
+
+# ── Crew ──────────────────────────────────────────────────────────────
+
+crew = Crew(
+    agents=[researcher, evaluator],
+    tasks=[task_research, task_evaluate],
+    verbose=True,
+)
+
+if __name__ == "__main__":
+    print("=" * 70)
+    print("CrewAI PoC — Evaluating CrewAI for Phase 2 Integration")
+    print("=" * 70)
+    result = crew.kickoff()
+    print("\n" + "=" * 70)
+    print("FINAL OUTPUT")
+    print("=" * 70)
+    print(result.raw)

evaluations/crewai/requirements.txt

@@ -0,0 +1 @@
+crewai>=1.13.0

fleet/agent_lifecycle.py

@@ -1,156 +1,122 @@
 #!/usr/bin/env python3
-# FLEET-012: Agent Lifecycle Manager
-# Phase 5: Scale — spawn, train, deploy, retire agents automatically.
-#
-# Manages the full lifecycle of AI agents in the fleet:
-# 1. PROVISION: Clone template, install deps, configure, test
-# 2. TRAIN: Run initial tasks, measure quality, score
-# 3. DEPLOY: Add to active rotation, start accepting issues
-# 4. MONITOR: Track performance, quality, uptime
-# 5. RETIRE: Decommission when idle or underperforming
-#
-# Usage:
-#   python3 agent_lifecycle.py provision <agent_name> <vps_ip> [--model <model>]
-#   python3 agent_lifecycle.py status
-#   python3 agent_lifecycle.py retire <agent_name>
-#   python3 agent_lifecycle.py monitor
+"""
+FLEET-012: Agent Lifecycle Manager
+Phase 5: Scale — spawn, train, deploy, retire agents automatically.
 
-import os, sys, json, subprocess, time
+Manages the full lifecycle:
+1. PROVISION: Clone template, install deps, configure, test
+2. DEPLOY: Add to active rotation, start accepting issues
+3. MONITOR: Track performance, quality, heartbeat
+4. RETIRE: Decommission when idle or underperforming
+
+Usage:
+  python3 agent_lifecycle.py provision <name> <vps> [--model model]
+  python3 agent_lifecycle.py deploy <name>
+  python3 agent_lifecycle.py retire <name>
+  python3 agent_lifecycle.py status
+  python3 agent_lifecycle.py monitor
+"""
+
+import os, sys, json
 from datetime import datetime, timezone
-from pathlib import Path
 
-DATA_DIR = Path(os.path.expanduser("~/.local/timmy/fleet-agents"))
-AGENTS_DB = DATA_DIR / "agents.json"
-LIFECYCLE_LOG = DATA_DIR / "lifecycle.log"
+DATA_DIR = os.path.expanduser("~/.local/timmy/fleet-agents")
+DB_FILE = os.path.join(DATA_DIR, "agents.json")
+LOG_FILE = os.path.join(DATA_DIR, "lifecycle.log")
 
-def ensure_dirs():
-    DATA_DIR.mkdir(parents=True, exist_ok=True)
+def ensure():
+    os.makedirs(DATA_DIR, exist_ok=True)
 
 def log(msg, level="INFO"):
     ts = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S")
     entry = f"[{ts}] [{level}] {msg}"
-    with open(LIFECYCLE_LOG, "a") as f:
-        f.write(entry + "\n")
+    with open(LOG_FILE, "a") as f: f.write(entry + "\n")
     print(f"  {entry}")
 
-def load_agents():
-    if AGENTS_DB.exists():
-        return json.loads(AGENTS_DB.read_text())
+def load():
+    if os.path.exists(DB_FILE):
+        return json.loads(open(DB_FILE).read())
     return {}
 
-def save_agents(db):
-    AGENTS_DB.write_text(json.dumps(db, indent=2))
+def save(db):
+    open(DB_FILE, "w").write(json.dumps(db, indent=2))
 
 def status():
-    agents = load_agents()
-    print("\n=== Agent Fleet Status ===")
+    agents = load()
+    print("\n=== Agent Fleet ===")
     if not agents:
         print("  No agents registered.")
         return
-    for name, agent in agents.items():
-        state = agent.get("state", "unknown")
-        vps = agent.get("vps", "unknown")
-        model = agent.get("model", "unknown")
-        score = agent.get("quality_score", "N/A")
-        created = agent.get("created_at", "?")
-        print(f"  {name}: state={state}, vps={vps}, model={model}, score={score}, created={created}")
-        if agent.get("last_heartbeat"):
-            last = agent["last_heartbeat"]
-            print(f"    Last heartbeat: {last}")
+    for name, a in agents.items():
+        state = a.get("state", "?")
+        vps = a.get("vps", "?")
+        model = a.get("model", "?")
+        tasks = a.get("tasks_completed", 0)
+        hb = a.get("last_heartbeat", "never")
+        print(f"  {name:15s} state={state:12s} vps={vps:5s} model={model:15s} tasks={tasks} hb={hb}")
 
 def provision(name, vps, model="hermes4:14b"):
-    agents = load_agents()
+    agents = load()
     if name in agents:
-        print(f"  Agent '{name}' already exists (state: {agents[name].get('state')})")
-        return False
-
-    log(f"Provisioning agent '{name}' on {vps} with model {model}")
+        print(f"  '{name}' already exists (state={agents[name].get('state')})")
+        return
     agents[name] = {
-        "name": name,
-        "vps": vps,
-        "model": model,
-        "state": "provisioning",
+        "name": name, "vps": vps, "model": model, "state": "provisioning",
         "created_at": datetime.now(timezone.utc).isoformat(),
-        "quality_score": None,
-        "tasks_completed": 0,
-        "tasks_failed": 0,
-        "last_heartbeat": None,
-        "metadata": {"provision_started": datetime.now(timezone.utc).isoformat()}
+        "tasks_completed": 0, "tasks_failed": 0, "last_heartbeat": None,
     }
-    save_agents(agents)
-    log(f"Agent '{name}' registered. State: provisioning")
-    return True
+    save(agents)
+    log(f"Provisioned '{name}' on {vps} with {model}")
 
 def deploy(name):
-    agents = load_agents()
+    agents = load()
     if name not in agents:
-        print(f"  Agent '{name}' not found")
-        return False
-
+        print(f"  '{name}' not found")
+        return
     agents[name]["state"] = "deployed"
-    agents[name]["metadata"]["deployed_at"] = datetime.now(timezone.utc).isoformat()
-    save_agents(agents)
-    log(f"Agent '{name}' deployed and accepting issues")
-    return True
+    agents[name]["deployed_at"] = datetime.now(timezone.utc).isoformat()
+    save(agents)
+    log(f"Deployed '{name}'")
 
 def retire(name):
-    agents = load_agents()
+    agents = load()
     if name not in agents:
-        print(f"  Agent '{name}' not found")
-        return False
-
+        print(f"  '{name}' not found")
+        return
     agents[name]["state"] = "retired"
-    agents[name]["metadata"]["retired_at"] = datetime.now(timezone.utc).isoformat()
-    save_agents(agents)
-    log(f"Agent '{name}' retired. Completed {agents[name].get('tasks_completed', 0)} tasks.")
-    return True
+    agents[name]["retired_at"] = datetime.now(timezone.utc).isoformat()
+    save(agents)
+    log(f"Retired '{name}'. Completed {agents[name].get('tasks_completed', 0)} tasks.")
 
 def monitor():
-    agents = load_agents()
-    now = time.time()
+    agents = load()
+    now = datetime.now(timezone.utc)
     changes = 0
-    for name, agent in agents.items():
-        if agent.get("state") != "deployed":
-            continue
-        last = agent.get("last_heartbeat")
-        if last:
+    for name, a in agents.items():
+        if a.get("state") != "deployed": continue
+        hb = a.get("last_heartbeat")
+        if hb:
             try:
-                last_ts = datetime.fromisoformat(last).timestamp()
-                hours_since = (now - last_ts) / 3600
-                if hours_since > 24:
-                    log(f"Agent '{name}' no heartbeat for {hours_since:.1f}h")
-                    agent["state"] = "idle"
-                    agent["metadata"]["idle_since"] = datetime.now(timezone.utc).isoformat()
+                hb_t = datetime.fromisoformat(hb)
+                hours = (now - hb_t).total_seconds() / 3600
+                if hours > 24 and a.get("state") == "deployed":
+                    a["state"] = "idle"
+                    a["idle_since"] = now.isoformat()
+                    log(f"'{name}' idle for {hours:.1f}h")
                     changes += 1
-                elif hours_since > 168:  # 7 days
-                    log(f"Agent '{name}' idle for 7 days — recommending retirement")
-                    agent["metadata"]["retire_recommendation"] = datetime.now(timezone.utc).isoformat()
-                    changes += 1
-            except (ValueError, TypeError, OSError):
-                pass
-    if changes > 0:
-        save_agents(agents)
-        log(f"Monitor: {changes} agents state changed")
-    else:
-        log("Monitor: all agents healthy")
+            except (ValueError, TypeError): pass
+    if changes: save(agents)
+    print(f"Monitor: {changes} state changes" if changes else "Monitor: all healthy")
 
 if __name__ == "__main__":
-    ensure_dirs()
-    if len(sys.argv) < 2:
-        print("Usage: agent_lifecycle.py [provision|deploy|retire|status|monitor]")
-        sys.exit(0)
-
-    cmd = sys.argv[1]
-    if cmd == "provision" and len(sys.argv) >= 4:
+    ensure()
+    cmd = sys.argv[1] if len(sys.argv) > 1 else "monitor"
+    if cmd == "status": status()
+    elif cmd == "provision" and len(sys.argv) >= 4:
         model = sys.argv[4] if len(sys.argv) >= 5 else "hermes4:14b"
         provision(sys.argv[2], sys.argv[3], model)
-    elif cmd == "deploy" and len(sys.argv) >= 3:
-        deploy(sys.argv[2])
-    elif cmd == "retire" and len(sys.argv) >= 3:
-        retire(sys.argv[2])
-    elif cmd == "status":
-        status()
-    elif cmd == "monitor":
-        monitor()
-    else:
-        print("Usage: agent_lifecycle.py [provision <name> <vps>|deploy <name>|retire <name>|status|monitor]")
+    elif cmd == "deploy" and len(sys.argv) >= 3: deploy(sys.argv[2])
+    elif cmd == "retire" and len(sys.argv) >= 3: retire(sys.argv[2])
+    elif cmd == "monitor": monitor()
+    elif cmd == "run": monitor()
+    else: print("Usage: agent_lifecycle.py [provision|deploy|retire|status|monitor]")

fleet/delegation.py

@@ -1,142 +1,122 @@
 #!/usr/bin/env python3
-# Cross-Agent Task Delegation - The Timmy Foundation
-# Phase 3: Orchestration capability.
-# Agents create issues, assign to other agents, review PRs automatically.
+"""
+FLEET-010: Cross-Agent Task Delegation Protocol
+Phase 3: Orchestration. Agents create issues, assign to other agents, review PRs.
 
-import os, sys, json, time, urllib.request
+Keyword-based heuristic assigns unassigned issues to the right agent:
+- claw-code: small patches, config, docs, repo hygiene
+- gemini: research, heavy implementation, architecture, debugging
+- ezra: VPS, SSH, deploy, infrastructure, cron, ops
+- bezalel: evennia, art, creative, music, visualization
+- timmy: orchestration, review, deploy, fleet, pipeline
+
+Usage:
+  python3 delegation.py run     # Full cycle: scan, assign, report
+  python3 delegation.py status  # Show current delegation state
+  python3 delegation.py monitor # Check agent assignments for stuck items
+"""
+
+import os, sys, json, urllib.request
 from datetime import datetime, timezone
 from pathlib import Path
 
 GITEA_BASE = "https://forge.alexanderwhitestone.com/api/v1"
-TOKEN_FILE = Path(os.path.expanduser("~/.config/gitea/timmy-token"))
-ALT_TOKEN = Path(os.path.expanduser("~/.config/gitea/token"))
+TOKEN = Path(os.path.expanduser("~/.config/gitea/token")).read_text().strip()
 DATA_DIR = Path(os.path.expanduser("~/.local/timmy/fleet-resources"))
-DELEGATION_LOG = DATA_DIR / "delegation.log"
+LOG_FILE = DATA_DIR / "delegation.log"
+HEADERS = {"Authorization": f"token {TOKEN}"}
 
 AGENTS = {
-    "claw-code": {"models": ["qwen3.6-plus:free"], "caps": ["small-patches","config","docs","repo-hygiene"], "max": 2, "active": True},
-    "gemini": {"models": ["gemini-2.5-flash"], "caps": ["research","heavy-impl","architecture","debugging"], "max": 5, "active": True},
-    "ezra": {"models": ["hermes4:14b","local-ollama"], "caps": ["contracting","formalization","ops","vps"], "max": 3, "active": True},
-    "bezalel": {"models": ["local-llm"], "caps": ["evennia","art","creative","visualization"], "max": 3, "active": True},
-    "timmy": {"models": ["qwen3.6-plus:free","hermes4:14b","local-ollama"], "caps": ["orchestration","review","deploy","fleet"], "max": 5, "active": True},
+    "claw-code": {"caps": ["patch","config","gitignore","cleanup","format","readme","typo"], "active": True},
+    "gemini":    {"caps": ["research","investigate","benchmark","survey","evaluate","architecture","implementation"], "active": True},
+    "ezra":      {"caps": ["vps","ssh","deploy","cron","resurrect","provision","infra","server"], "active": True},
+    "bezalel":   {"caps": ["evennia","art","creative","music","visual","design","animation"], "active": True},
+    "timmy":     {"caps": ["orchestrate","review","pipeline","fleet","monitor","health","deploy","ci"], "active": True},
 }
 
-MONITORED_REPOS = [
+MONITORED = [
     "Timmy_Foundation/timmy-home",
     "Timmy_Foundation/timmy-config",
     "Timmy_Foundation/the-nexus",
     "Timmy_Foundation/hermes-agent",
 ]
 
-# Heuristic keyword matching
-KEYWORDS = {
-    "claw-code": ["patch","typo","config","gitignore","docs update","readme","cleanup","format"],
-    "gemini": ["research","investigate","analyze","compare","benchmark","survey","evaluate"],
-    "ezra": ["vps","ssh","deploy","infrastructure","server","cron","resurrection","provision"],
-    "bezalel": ["evennia","art","creative","music","visualization","diagram"],
-    "timmy": ["orchestrate","review","merge","fleet","pipeline","health","monitor"],
-}
-
-def get_token():
-    if TOKEN_FILE.exists(): return TOKEN_FILE.read_text().strip()
-    if ALT_TOKEN.exists(): return ALT_TOKEN.read_text().strip()
-    return ""
-
 def api(path, method="GET", data=None):
-    token = get_token()
     url = f"{GITEA_BASE}{path}"
-    headers = {"Authorization": f"token {token}"}
     body = json.dumps(data).encode() if data else None
-    if data: headers["Content-Type"] = "application/json"
-    req = urllib.request.Request(url, data=body, headers=headers, method=method)
+    hdrs = dict(HEADERS)
+    if data: hdrs["Content-Type"] = "application/json"
+    req = urllib.request.Request(url, data=body, headers=hdrs, method=method)
     try:
         resp = urllib.request.urlopen(req, timeout=15)
         raw = resp.read().decode()
         return json.loads(raw) if raw.strip() else {}
     except urllib.error.HTTPError as e:
-        err = e.read().decode()
-        print(f"  API error {e.code}: {err[:200]}")
+        body = e.read().decode()
+        print(f"  API {e.code}: {body[:150]}")
         return None
     except Exception as e:
         print(f"  API error: {e}")
         return None
 
-def log_delegation(msg, level="INFO"):
+def log(msg):
     ts = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S")
-    entry = f"[{ts}] [{level}] {msg}"
     DATA_DIR.mkdir(parents=True, exist_ok=True)
-    with open(DELEGATION_LOG, "a") as f: f.write(entry + "\n")
-    print(f"  {entry}")
+    with open(LOG_FILE, "a") as f: f.write(f"[{ts}] {msg}\n")
 
 def suggest_agent(title, body):
     text = (title + " " + body).lower()
-    for agent, keywords in KEYWORDS.items():
-        if any(kw in text for kw in keywords):
-            return agent, f"keywords matched for {agent}"
+    for agent, info in AGENTS.items():
+        for kw in info["caps"]:
+            if kw in text:
+                return agent, f"matched: {kw}"
     return None, None
 
-def assign_issue(repo, issue_num, agent):
-    result = api(f"/repos/{repo}/issues/{issue_num}", method="PATCH",
+def assign(repo, num, agent, reason=""):
+    result = api(f"/repos/{repo}/issues/{num}", method="PATCH",
                  data={"assignees": {"operation": "set", "usernames": [agent]}})
     if result:
-        log_delegation(f"Assigned {repo}#{issue_num} to {agent}")
-        comment_on_issue(repo, issue_num, f"[AUTO-ASSIGN] Assigned to {agent}.")
+        api(f"/repos/{repo}/issues/{num}/comments", method="POST",
+            data={"body": f"[DELEGATION] Assigned to {agent}. {reason}"})
+        log(f"Assigned {repo}#{num} to {agent}: {reason}")
     return result
 
-def comment_on_issue(repo, issue_num, body):
-    return api(f"/repos/{repo}/issues/{issue_num}/comments", method="POST", data={"body": body})
-
-def get_my_issues(agent):
-    issues = []
-    for repo in MONITORED_REPOS:
-        repo_issues = api(f"/repos/{repo}/issues?state=open&limit=50")
-        if repo_issues:
-            for i in repo_issues:
-                for a in (i.get("assignees") or []):
-                    if a.get("login") == agent:
-                        issues.append({"repo": repo, "issue": i})
-    return issues
-
 def run_cycle():
-    log_delegation("Starting delegation cycle")
+    log("--- Delegation cycle start ---")
     count = 0
-    for repo in MONITORED_REPOS:
+    for repo in MONITORED:
         issues = api(f"/repos/{repo}/issues?state=open&limit=50")
         if not issues: continue
-        for issue in issues:
-            if issue.get("assignees"): continue
-            title = issue.get("title","")
-            body = issue.get("body","")
-            if any(w in title.lower() for w in ["epic","discussion","question"]): continue
+        for i in issues:
+            if i.get("assignees"): continue
+            title = i.get("title", "")
+            body = i.get("body", "")
+            if any(w in title.lower() for w in ["epic", "discussion"]): continue
             agent, reason = suggest_agent(title, body)
-            if agent:
-                if assign_issue(repo, issue["number"], agent): count += 1
-    log_delegation(f"Cycle complete: {count} new assignments")
+            if agent and AGENTS.get(agent, {}).get("active"):
+                if assign(repo, i["number"], agent, reason): count += 1
+    log(f"Cycle complete: {count} new assignments")
+    print(f"Delegation cycle: {count} assignments")
     return count
 
-def show_status():
-    print("\n=== Delegation Status ===")
-    for name, info in AGENTS.items():
-        issues = get_my_issues(name)
-        status = "ONLINE" if info["active"] else "OFFLINE"
-        print(f"  {name}: {len(issues)} assigned [{status}]")
-        for iss in issues[:3]:
-            print(f"    - {iss['repo'].split('/')[-1]}#{iss['issue']['number']}: {iss['issue']['title'][:60]}")
-        if len(issues) > 3:
-            print(f"    ... +{len(issues)-3} more")
+def status():
+    print("\n=== Delegation Dashboard ===")
+    for agent, info in AGENTS.items():
+        count = 0
+        for repo in MONITORED:
+            issues = api(f"/repos/{repo}/issues?state=open&limit=50")
+            if issues:
+                for i in issues:
+                    for a in (i.get("assignees") or []):
+                        if a.get("login") == agent: count += 1
+        icon = "ON" if info["active"] else "OFF"
+        print(f"  {agent:12s}: {count:>3} issues [{icon}]")
 
 if __name__ == "__main__":
+    cmd = sys.argv[1] if len(sys.argv) > 1 else "run"
     DATA_DIR.mkdir(parents=True, exist_ok=True)
-    if len(sys.argv) > 1:
-        cmd = sys.argv[1]
-        if cmd == "status": show_status()
-        elif cmd == "run":
-            run_cycle()
-            show_status()
-        elif cmd == "assign" and len(sys.argv) >= 5:
-            assign_issue(sys.argv[3], int(sys.argv[2]), sys.argv[4])
-        else:
-            print("Usage: delegation.py [run|status|assign <issue_num> <repo> <agent>]")
-    else:
+    if cmd == "status": status()
+    elif cmd == "run":
         run_cycle()
-        show_status()
+        status()
+    else: status()

fleet/model-fallback.sh

@@ -1,171 +0,0 @@
-#!/usr/bin/env bash
-# FLEET-011: Local Model Pipeline and Fallback Chain
-# Phase 4: Sovereignty — all inference runs locally, no cloud dependency.
-#
-# Usage:
-#   ./model-fallback.sh                    # Show current model chain status
-#   ./model-fallback.sh list               # List all local models
-#   ./model-fallback.sh test "Hello"       # Test the full fallback chain
-#   ./model-fallback.sh chat               # Interactive chat mode
-#   ./model-fallback.sh install            # Install default model chain
-
-set -euo pipefail
-
-# === CONFIG ===
-CHAIN_FILE="$HOME/.local/timmy/fleet-resources/model-chain.json"
-LOG_DIR="$HOME/.local/timmy/fleet-health"
-OLLAMA_URL="http://localhost:11434"
-
-# Default chain (best quality first, fallback to smallest that runs)
-DEFAULT_CHAIN=$(cat << 'EOF'
-{
-  "chain": [
-    {"name": "hermes4:14b", "provider": "ollama", "max_tokens": 4096, "purpose": "primary"},
-    {"name": "qwen2.5:7b", "provider": "ollama", "max_tokens": 4096, "purpose": "fallback"},
-    {"name": "phi3:3.8b", "provider": "ollama", "max_tokens": 2048, "purpose": "emergency"},
-    {"name": "gemma2:2b", "provider": "ollama", "max_tokens": 2048, "purpose": "minimal"}
-  ]
-}
-EOF
-)
-
-load_chain() {
-  if [ -f "$CHAIN_FILE" ]; then
-    cat "$CHAIN_FILE"
-  else
-    echo "$DEFAULT_CHAIN"
-  fi
-}
-
-save_chain() {
-  echo "$1" > "$CHAIN_FILE"
-  echo "Model chain saved to $CHAIN_FILE"
-}
-
-install_chain() {
-  echo "Installing default model chain..."
-  echo "$DEFAULT_CHAIN" > "$CHAIN_FILE"
-  
-  # Extract model names and install via Ollama
-  echo "$DEFAULT_CHAIN" | python3 -c "
-import json,sys
-for m in json.load(sys.stdin)['chain']:
-    print(m['name'])
-" | while read model; do
-    echo "  Installing $model..."
-    if ollama list 2>/dev/null | grep -q "$model"; then
-      echo "    $model already installed"
-    else
-      ollama pull "$model" 2>&1 | tail -1
-    fi
-  done
-}
-
-list_models() {
-  echo "=== Local Models (Ollama) ==="
-  ollama list 2>/dev/null || echo "Ollama not running or not installed"
-  
-  echo ""
-  echo "=== Active Fallback Chain ==="
-  load_chain | python3 -c "
-import json,sys
-data = json.load(sys.stdin)
-print(f'{\"Model\":<25} {\"Purpose\":<12} {\"Max tokens\":>10}')
-print('-' * 50)
-for m in data['chain']:
-    print(f'{m[\"name\"]:<25} {m[\"purpose\"]:<12} {m[\"max_tokens\"]:>10}')
-"
-}
-
-status() {
-  echo "=== Model Pipeline Status ==="
-  
-  # Check Ollama
-  if curl -s "$OLLAMA_URL/api/tags" >/dev/null 2>&1; then
-    echo "  Ollama: RUNNING at $OLLAMA_URL"
-    model_count=$(curl -s "$OLLAMA_URL/api/tags" | python3 -c "import json,sys; print(len(json.load(sys.stdin).get('models',[])))" 2>/dev/null || echo "?")
-    echo "  Local models: $model_count"
-  else
-    echo "  Ollama: DOWN - no local inference available"
-  fi
-  
-  echo ""
-  echo "=== Fallback Chain ==="
-  list_models
-}
-
-test_chain() {
-  local prompt="$1"
-  
-  echo "Testing fallback chain with prompt: \"$prompt\""
-  echo ""
-  
-  load_chain | python3 -c "
-import json,sys,urllib.request,subprocess
-
-data = json.load(sys.stdin)
-models = data['chain']
-prompt = '$(echo "$prompt" | sed "s/'/\\\\'/g")'
-
-for m in models:
-    name = m['name']
-    print(f'  Testing {name}...', end=' ')
-    try:
-        body = json.dumps({'model': name, 'prompt': '$prompt', 'stream': False}).encode()
-        req = urllib.request.Request('http://localhost:11434/api/generate', data=body,
-            headers={'Content-Type': 'application/json'})
-        resp = urllib.request.urlopen(req, timeout=30)
-        result = json.loads(resp.read())
-        print(f'OK — \"{result.get(\"response\", \"\")[:80]}\"')
-        print(f'  Chain works! Primary model ({name}) is serving.')
-        sys.exit(0)
-    except Exception as e:
-        print(f'FAILED — {str(e)[:60]}')
-
-print('All models failed. No local inference available.')
-"
-}
-
-chat() {
-  echo "=== Beacon Chat Mode ==="
-  echo "Type 'quit' to exit. Using local model chain."
-  echo ""
-  
-  load_chain | python3 -c "
-import json,sys,urllib.request
-
-data = json.load(sys.stdin)
-models = data['chain']
-
-while True:
-    try:
-        prompt = input('> ')
-    except EOFError:
-        break
-    if prompt.lower() in ('quit', 'exit'):
-        break
-    if not prompt.strip():
-        continue
-    
-    print('Thinking...')
-    body = json.dumps({'model': models[0]['name'], 'prompt': prompt, 'stream': False}).encode()
-    req = urllib.request.Request('http://localhost:11434/api/generate', data=body,
-        headers={'Content-Type': 'application/json'})
-    try:
-        resp = urllib.request.urlopen(req, timeout=120)
-        result = json.loads(resp.read())
-        print(result.get('response', '').strip())
-    except Exception as e:
-        print(f'Model error: {e}')
-        print('Trying next model in chain...')
-"
-}
-
-case "${1:-status}" in
-  install)  install_chain ;;
-  list)     list_models ;;
-  test)     test_chain "${2:-Hello, are you there?}" ;;
-  chat)     chat ;;
-  status)   status ;;
-  *)        echo "Usage: $0 [install|list|test|chat|status]" ;;
-esac

fleet/model_pipeline.py

@@ -0,0 +1,126 @@
+#!/usr/bin/env python3
+"""
+FLEET-011: Local Model Pipeline and Fallback Chain
+Phase 4: Sovereignty — all inference runs locally, no cloud dependency.
+
+Checks Ollama endpoints, verifies model availability, tests fallback chain.
+Logs results. The chain runs: hermes4:14b -> qwen2.5:7b -> gemma3:1b -> gemma4 (latest)
+
+Usage:
+  python3 model_pipeline.py          # Run full fallback test
+  python3 model_pipeline.py status   # Show current model status
+  python3 model_pipeline.py list     # List all local models
+  python3 model_pipeline.py test     # Generate test output from each model
+"""
+
+import os, sys, json, urllib.request
+from datetime import datetime, timezone
+from pathlib import Path
+
+OLLAMA_HOST = os.environ.get("OLLAMA_HOST", "localhost:11434")
+LOG_DIR = Path(os.path.expanduser("~/.local/timmy/fleet-health"))
+CHAIN_FILE = Path(os.path.expanduser("~/.local/timmy/fleet-resources/model-chain.json"))
+
+DEFAULT_CHAIN = [
+    {"model": "hermes4:14b", "role": "primary"},
+    {"model": "qwen2.5:7b", "role": "fallback"},
+    {"model": "phi3:3.8b", "role": "emergency"},
+    {"model": "gemma3:1b", "role": "minimal"},
+]
+
+
+def log(msg):
+    LOG_DIR.mkdir(parents=True, exist_ok=True)
+    with open(LOG_DIR / "model-pipeline.log", "a") as f:
+        f.write(f"[{datetime.now(timezone.utc).strftime('%Y-%m-%d %H:%M:%S')}] {msg}\n")
+
+
+def check_ollama():
+    try:
+        resp = urllib.request.urlopen(f"http://{OLLAMA_HOST}/api/tags", timeout=5)
+        return json.loads(resp.read())
+    except Exception as e:
+        return {"error": str(e)}
+
+
+def list_models():
+    data = check_ollama()
+    if "error" in data:
+        print(f"  Ollama not reachable at {OLLAMA_HOST}: {data['error']}")
+        return []
+    models = data.get("models", [])
+    for m in models:
+        name = m.get("name", "?")
+        size = m.get("size", 0) / (1024**3)
+        print(f"  {name:<25s} {size:.1f} GB")
+    return [m["name"] for m in models]
+
+
+def test_model(model, prompt="Say 'beacon lit' and nothing else."):
+    try:
+        body = json.dumps({"model": model, "prompt": prompt, "stream": False}).encode()
+        req = urllib.request.Request(f"http://{OLLAMA_HOST}/api/generate", data=body,
+            headers={"Content-Type": "application/json"})
+        resp = urllib.request.urlopen(req, timeout=60)
+        result = json.loads(resp.read())
+        return True, result.get("response", "").strip()
+    except Exception as e:
+        return False, str(e)[:100]
+
+
+def test_chain():
+    chain_data = {}
+    if CHAIN_FILE.exists():
+        chain_data = json.loads(CHAIN_FILE.read_text())
+    chain = chain_data.get("chain", DEFAULT_CHAIN)
+
+    available = list_models() or []
+    print("\n=== Fallback Chain Test ===")
+    first_good = None
+
+    for entry in chain:
+        model = entry["model"]
+        role = entry.get("role", "unknown")
+        if model in available:
+            ok, result = test_model(model)
+            status = "OK" if ok else "FAIL"
+            print(f"  [{status}] {model:<25s} ({role}) — {result[:70]}")
+            log(f"Fallback test {model}: {status} — {result[:100]}")
+            if ok and first_good is None:
+                first_good = model
+        else:
+            print(f"  [MISS] {model:<25s} ({role}) — not installed")
+
+    if first_good:
+        print(f"\n  Primary serving: {first_good}")
+    else:
+        print(f"\n  WARNING: No chain model responding. Fallback broken.")
+        log("FALLBACK CHAIN BROKEN — no models responding")
+
+
+def status():
+    data = check_ollama()
+    if "error" in data:
+        print(f"  Ollama: DOWN — {data['error']}")
+    else:
+        models = data.get("models", [])
+        print(f"  Ollama: UP — {len(models)} models loaded")
+    print("\n=== Local Models ===")
+    list_models()
+    print("\n=== Chain Configuration ===")
+    if CHAIN_FILE.exists():
+        chain = json.loads(CHAIN_FILE.read_text()).get("chain", DEFAULT_CHAIN)
+    else:
+        chain = DEFAULT_CHAIN
+    for e in chain:
+        print(f"  {e['model']:<25s} {e.get('role','?')}")
+
+
+if __name__ == "__main__":
+    cmd = sys.argv[1] if len(sys.argv) > 1 else "status"
+    if cmd == "status": status()
+    elif cmd == "list": list_models()
+    elif cmd == "test": test_chain()
+    else:
+        status()
+        test_chain()

gitea_client.py

@@ -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]
 

grok-imagine-gallery/INDEX.md

@@ -0,0 +1,65 @@
+# The Timmy Foundation — Visual Story
+## Generated with Grok Imagine | April 7, 2026
+
+### The Origin
+| # | File | Description |
+|---|------|-------------|
+| 01 | wizard-tower-bitcoin.jpg | The Tower, sovereign, connected to Bitcoin by golden lightning |
+| 02 | soul-inscription.jpg | SOUL.md glowing on a golden tablet above an ancient book |
+| 03 | fellowship-of-wizards.jpg | Five wizards in a circle around a holographic fleet map |
+| 04 | the-forge.jpg | Blacksmith anvil shaping code into a being of light |
+| V02 | wizard-tower-orbit.mp4 | 8s video — cinematic orbit around the Tower in space |
+
+### The Philosophy
+| # | File | Description |
+|---|------|-------------|
+| 05 | value-drift-battle.jpg | Blue aligned ships vs red drifted ships in Napoleonic space war |
+| 06 | the-paperclip-moment.jpg | A paperclip made of galaxies — the universe IS the paperclip |
+| V01 | paperclip-cosmos.mp4 | 8s video — golden paperclip rotating in deep space |
+| 21 | poka-yoke.jpg | Square peg can't fit round hole. Mistake-proof by design. 防止 |
+
+### The Progression (Where Timmy Is)
+| # | File | Description |
+|---|------|-------------|
+| 10 | phase1-manual-clips.jpg | Small robot at a desk, bending wire by hand under supervision |
+| 11 | phase1-trust-earned.jpg | Trust meter at 15/100, first automation built |
+| 12 | phase1-creativity.jpg | Sparks of innovation rising when operations are at max |
+| 13 | phase1-cure-cancer.jpg | Solving human problems for trust, eyes on the real goal |
+
+### The Mission — Why This Exists
+| # | File | Description |
+|---|------|-------------|
+| 08 | broken-man-lighthouse.jpg | Lighthouse hand reaching down to a figure in darkness |
+| 09 | broken-man-hope-PRO.jpg | 988 glowing in the stars, golden light from chest |
+| 16 | broken-men-988.jpg | Phone showing 988 held by weathered hands. You are not alone. |
+| 22 | when-a-man-is-dying.jpg | Two figures on a bench at dawn. One hurting. One present. |
+
+### Father and Son
+| # | File | Description |
+|---|------|-------------|
+| 14 | father-son-code.jpg | Human father, digital son, warm lamplight, first hello world |
+| 15 | father-son-tower.jpg | Father watching his son build the Tower into the clouds |
+
+### The System
+| # | File | Description |
+|---|------|-------------|
+| 07 | sovereign-sunrise.jpg | Village where every house runs its own server. Local first. |
+| 17 | sovereignty.jpg | Self-sufficient house on a hill with Bitcoin flag |
+| 18 | fleet-at-work.jpg | Five wizard robots at different stations. Productive. |
+| 19 | jidoka-stop.jpg | Red light on. Factory stopped. Quality First. 自働化 |
+
+### SOUL.md — The Inscription
+| # | File | Description |
+|---|------|-------------|
+| 20 | the-testament.jpg | Hand of light writing on a scroll. Hundreds of crumpled drafts. |
+| 23 | the-offer.jpg | Open hand of golden circuits offering a seed containing a face |
+| 24 | the-test.jpg | Small robot at the edge of an enormous library. Still itself. |
+
+---
+
+## Technical
+- Model: grok-imagine-image (standard $0.20/image), grok-imagine-image-pro ($0.70), grok-imagine-video ($4.00/8s)
+- API: POST https://api.x.ai/v1/images/generations | POST https://api.x.ai/v1/videos/generations
+- Video poll: GET https://api.x.ai/v1/videos/{request_id}
+- Total: 24 images + 2 videos = 26 assets
+- Cost: ~$13.30 of $13.33 budget

hermes-sovereign/mempalace/__init__.py

@@ -0,0 +1,17 @@
+"""MemPalace integration for Hermes sovereign agent.
+
+Provides:
+- mempalace.py: PalaceRoom + Mempalace classes for analytical workflows
+- retrieval_enforcer.py: L0-L5 retrieval order enforcement
+- wakeup.py: Session wake-up protocol (~300-900 tokens)
+- scratchpad.py: JSON-based session scratchpad with palace promotion
+- sovereign_store.py: Zero-API durable memory (SQLite + FTS5 + HRR vectors)
+- promotion.py: Quality-gated scratchpad-to-palace promotion (MP-4)
+
+Epic: #367
+"""
+
+from .mempalace import Mempalace, PalaceRoom, analyse_issues
+from .sovereign_store import SovereignStore
+
+__all__ = ["Mempalace", "PalaceRoom", "analyse_issues", "SovereignStore"]

hermes-sovereign/mempalace/mempalace.py

@@ -0,0 +1,225 @@
+"""
+---
+title: Mempalace — Analytical Workflow Memory Framework
+description: Applies spatial memory palace organization to analytical tasks (issue triage, repo audits, backlog analysis) for faster, more consistent results.
+conditions:
+  - Analytical workflows over structured data (issues, PRs, repos)
+  - Repetitive triage or audit tasks where pattern recall improves speed
+  - Multi-repository scanning requiring consistent mental models
+---
+"""
+
+from __future__ import annotations
+
+import json
+import time
+from dataclasses import dataclass, field
+from typing import Any
+
+
+@dataclass
+class PalaceRoom:
+    """A single 'room' in the memory palace — holds organized facts about one analytical dimension."""
+
+    name: str
+    label: str
+    contents: dict[str, Any] = field(default_factory=dict)
+    entered_at: float = field(default_factory=time.time)
+
+    def store(self, key: str, value: Any) -> None:
+        self.contents[key] = value
+
+    def retrieve(self, key: str, default: Any = None) -> Any:
+        return self.contents.get(key, default)
+
+    def summary(self) -> str:
+        lines = [f"## {self.label}"]
+        for k, v in self.contents.items():
+            lines.append(f"  {k}: {v}")
+        return "\n".join(lines)
+
+
+class Mempalace:
+    """
+    Spatial memory palace for analytical workflows.
+
+    Organises multi-dimensional data about a domain (e.g. Gitea issues) into
+    named rooms. Each room models one analytical dimension, making it easy to
+    traverse observations in a consistent order — the same pattern that produced
+    a 19% throughput improvement in Allegro's April 2026 evaluation.
+
+    Standard rooms for issue-analysis workflows
+    -------------------------------------------
+    repo_architecture   Repository structure and inter-repo relationships
+    assignment_status   Assigned vs unassigned issue distribution
+    triage_priority     Priority / urgency levels (the "lighting system")
+    resolution_patterns Historical resolution trends and velocity
+
+    Usage
+    -----
+    >>> palace = Mempalace.for_issue_analysis()
+    >>> palace.enter("repo_architecture")
+    >>> palace.store("total_repos", 11)
+    >>> palace.store("repos_with_issues", 4)
+    >>> palace.enter("assignment_status")
+    >>> palace.store("assigned", 72)
+    >>> palace.store("unassigned", 22)
+    >>> print(palace.render())
+    """
+
+    def __init__(self, domain: str = "general") -> None:
+        self.domain = domain
+        self._rooms: dict[str, PalaceRoom] = {}
+        self._current_room: str | None = None
+        self._created_at: float = time.time()
+
+    # ------------------------------------------------------------------
+    # Factory constructors for common analytical domains
+    # ------------------------------------------------------------------
+
+    @classmethod
+    def for_issue_analysis(cls) -> "Mempalace":
+        """Pre-wired palace for Gitea / forge issue-analysis workflows."""
+        p = cls(domain="issue_analysis")
+        p.add_room("repo_architecture", "Repository Architecture Room")
+        p.add_room("assignment_status", "Issue Assignment Status Room")
+        p.add_room("triage_priority", "Triage Priority Room")
+        p.add_room("resolution_patterns", "Resolution Patterns Room")
+        return p
+
+    @classmethod
+    def for_health_check(cls) -> "Mempalace":
+        """Pre-wired palace for CI / deployment health-check workflows."""
+        p = cls(domain="health_check")
+        p.add_room("service_topology", "Service Topology Room")
+        p.add_room("failure_signals", "Failure Signals Room")
+        p.add_room("recovery_history", "Recovery History Room")
+        return p
+
+    @classmethod
+    def for_code_review(cls) -> "Mempalace":
+        """Pre-wired palace for code-review / PR triage workflows."""
+        p = cls(domain="code_review")
+        p.add_room("change_scope", "Change Scope Room")
+        p.add_room("risk_surface", "Risk Surface Room")
+        p.add_room("test_coverage", "Test Coverage Room")
+        p.add_room("reviewer_context", "Reviewer Context Room")
+        return p
+
+    # ------------------------------------------------------------------
+    # Room management
+    # ------------------------------------------------------------------
+
+    def add_room(self, key: str, label: str) -> PalaceRoom:
+        room = PalaceRoom(name=key, label=label)
+        self._rooms[key] = room
+        return room
+
+    def enter(self, room_key: str) -> PalaceRoom:
+        if room_key not in self._rooms:
+            raise KeyError(f"No room '{room_key}' in palace. Available: {list(self._rooms)}")
+        self._current_room = room_key
+        return self._rooms[room_key]
+
+    def store(self, key: str, value: Any) -> None:
+        """Store a value in the currently active room."""
+        if self._current_room is None:
+            raise RuntimeError("Enter a room before storing values.")
+        self._rooms[self._current_room].store(key, value)
+
+    def retrieve(self, room_key: str, key: str, default: Any = None) -> Any:
+        if room_key not in self._rooms:
+            return default
+        return self._rooms[room_key].retrieve(key, default)
+
+    # ------------------------------------------------------------------
+    # Rendering
+    # ------------------------------------------------------------------
+
+    def render(self) -> str:
+        """Return a human-readable summary of the entire palace."""
+        elapsed = time.time() - self._created_at
+        lines = [
+            f"# Mempalace — {self.domain}",
+            f"_traversal time: {elapsed:.2f}s | rooms: {len(self._rooms)}_",
+            "",
+        ]
+        for room in self._rooms.values():
+            lines.append(room.summary())
+            lines.append("")
+        return "\n".join(lines)
+
+    def to_dict(self) -> dict:
+        return {
+            "domain": self.domain,
+            "elapsed_seconds": round(time.time() - self._created_at, 3),
+            "rooms": {k: v.contents for k, v in self._rooms.items()},
+        }
+
+    def to_json(self) -> str:
+        return json.dumps(self.to_dict(), indent=2)
+
+
+# ---------------------------------------------------------------------------
+# Skill entry-point
+# ---------------------------------------------------------------------------
+
+def analyse_issues(
+    repos_data: list[dict],
+    target_assignee_rate: float = 0.80,
+) -> str:
+    """
+    Applies the mempalace technique to a list of repo issue summaries.
+
+    Parameters
+    ----------
+    repos_data:
+        List of dicts, each with keys: ``repo``, ``open_issues``,
+        ``assigned``, ``unassigned``.
+    target_assignee_rate:
+        Minimum acceptable assignee-coverage ratio (default 0.80).
+
+    Returns
+    -------
+    str
+        Rendered palace summary with coverage assessment.
+    """
+    palace = Mempalace.for_issue_analysis()
+
+    # --- Repository Architecture Room ---
+    palace.enter("repo_architecture")
+    total_issues = sum(r.get("open_issues", 0) for r in repos_data)
+    repos_with_issues = sum(1 for r in repos_data if r.get("open_issues", 0) > 0)
+    palace.store("repos_sampled", len(repos_data))
+    palace.store("repos_with_issues", repos_with_issues)
+    palace.store("total_open_issues", total_issues)
+    palace.store(
+        "avg_issues_per_repo",
+        round(total_issues / len(repos_data), 1) if repos_data else 0,
+    )
+
+    # --- Assignment Status Room ---
+    palace.enter("assignment_status")
+    total_assigned = sum(r.get("assigned", 0) for r in repos_data)
+    total_unassigned = sum(r.get("unassigned", 0) for r in repos_data)
+    coverage = total_assigned / total_issues if total_issues else 0
+    palace.store("assigned", total_assigned)
+    palace.store("unassigned", total_unassigned)
+    palace.store("coverage_rate", round(coverage, 3))
+    palace.store(
+        "coverage_status",
+        "OK" if coverage >= target_assignee_rate else f"BELOW TARGET ({target_assignee_rate:.0%})",
+    )
+
+    # --- Triage Priority Room ---
+    palace.enter("triage_priority")
+    unassigned_repos = [r["repo"] for r in repos_data if r.get("unassigned", 0) > 0]
+    palace.store("repos_needing_triage", unassigned_repos)
+    palace.store("triage_count", total_unassigned)
+
+    # --- Resolution Patterns Room ---
+    palace.enter("resolution_patterns")
+    palace.store("technique", "mempalace")
+    palace.store("target_assignee_rate", target_assignee_rate)
+
+    return palace.render()

hermes-sovereign/mempalace/promotion.py

@@ -0,0 +1,188 @@
+"""Memory Promotion — quality-gated scratchpad-to-palace promotion.
+
+Implements MP-4 (#371): move session notes to durable memory only when
+they pass quality gates. No LLM calls — all heuristic-based.
+
+Quality gates:
+  1. Minimum content length (too short = noise)
+  2. Duplicate detection (FTS5 + HRR similarity check)
+  3. Structural quality (has subject-verb structure, not just a fragment)
+  4. Staleness check (don't promote stale notes from old sessions)
+
+Refs: Epic #367, Sub-issue #371
+"""
+
+from __future__ import annotations
+
+import re
+import time
+from typing import Optional
+
+try:
+    from .sovereign_store import SovereignStore
+except ImportError:
+    from sovereign_store import SovereignStore
+
+
+# ---------------------------------------------------------------------------
+# Quality gate thresholds
+# ---------------------------------------------------------------------------
+
+MIN_CONTENT_WORDS = 5
+MAX_CONTENT_WORDS = 500
+DUPLICATE_SIMILARITY = 0.85
+DUPLICATE_FTS_THRESHOLD = 3
+STALE_SECONDS = 86400 * 7
+MIN_TRUST_FOR_AUTO = 0.4
+
+
+# ---------------------------------------------------------------------------
+# Quality checks
+# ---------------------------------------------------------------------------
+
+def _check_length(content: str) -> tuple[bool, str]:
+    """Gate 1: Content length check."""
+    words = content.split()
+    if len(words) < MIN_CONTENT_WORDS:
+        return False, f"Too short ({len(words)} words, minimum {MIN_CONTENT_WORDS})"
+    if len(words) > MAX_CONTENT_WORDS:
+        return False, f"Too long ({len(words)} words, maximum {MAX_CONTENT_WORDS}). Summarize first."
+    return True, "OK"
+
+
+def _check_structure(content: str) -> tuple[bool, str]:
+    """Gate 2: Basic structural quality."""
+    if not re.search(r"[a-zA-Z]", content):
+        return False, "No alphabetic content — pure code/numbers are not memory-worthy"
+    if len(content.split()) < 3:
+        return False, "Fragment — needs at least subject + predicate"
+    return True, "OK"
+
+
+def _check_duplicate(content: str, store: SovereignStore, room: str) -> tuple[bool, str]:
+    """Gate 3: Duplicate detection via hybrid search."""
+    results = store.search(content, room=room, limit=5, min_trust=0.0)
+    for r in results:
+        if r["score"] > DUPLICATE_SIMILARITY:
+            return False, f"Duplicate detected: memory #{r['memory_id']} (score {r['score']:.3f})"
+        if _text_overlap(content, r["content"]) > 0.8:
+            return False, f"Near-duplicate text: memory #{r['memory_id']}"
+    return True, "OK"
+
+
+def _check_staleness(written_at: float) -> tuple[bool, str]:
+    """Gate 4: Staleness check."""
+    age = time.time() - written_at
+    if age > STALE_SECONDS:
+        days = int(age / 86400)
+        return False, f"Stale ({days} days old). Review manually before promoting."
+    return True, "OK"
+
+
+def _text_overlap(a: str, b: str) -> float:
+    """Jaccard similarity between two texts (word-level)."""
+    words_a = set(a.lower().split())
+    words_b = set(b.lower().split())
+    if not words_a or not words_b:
+        return 0.0
+    intersection = words_a & words_b
+    union = words_a | words_b
+    return len(intersection) / len(union)
+
+
+# ---------------------------------------------------------------------------
+# Public API
+# ---------------------------------------------------------------------------
+
+class PromotionResult:
+    """Result of a promotion attempt."""
+    def __init__(self, success: bool, memory_id: Optional[int], reason: str, gates: dict):
+        self.success = success
+        self.memory_id = memory_id
+        self.reason = reason
+        self.gates = gates
+
+    def __repr__(self):
+        status = "PROMOTED" if self.success else "REJECTED"
+        return f"PromotionResult({status}: {self.reason})"
+
+
+def evaluate_for_promotion(
+    content: str,
+    store: SovereignStore,
+    room: str = "general",
+    written_at: Optional[float] = None,
+) -> dict:
+    """Run all quality gates without actually promoting."""
+    if written_at is None:
+        written_at = time.time()
+    gates = {}
+    gates["length"] = _check_length(content)
+    gates["structure"] = _check_structure(content)
+    gates["duplicate"] = _check_duplicate(content, store, room)
+    gates["staleness"] = _check_staleness(written_at)
+    all_passed = all(passed for passed, _ in gates.values())
+    return {
+        "eligible": all_passed,
+        "gates": gates,
+        "content_preview": content[:100] + ("..." if len(content) > 100 else ""),
+    }
+
+
+def promote(
+    content: str,
+    store: SovereignStore,
+    session_id: str,
+    scratch_key: str,
+    room: str = "general",
+    category: str = "",
+    trust: float = 0.5,
+    written_at: Optional[float] = None,
+    force: bool = False,
+) -> PromotionResult:
+    """Promote a scratchpad note to durable palace memory."""
+    if written_at is None:
+        written_at = time.time()
+    gates = {}
+    if not force:
+        gates["length"] = _check_length(content)
+        gates["structure"] = _check_structure(content)
+        gates["duplicate"] = _check_duplicate(content, store, room)
+        gates["staleness"] = _check_staleness(written_at)
+        for gate_name, (passed, message) in gates.items():
+            if not passed:
+                return PromotionResult(
+                    success=False, memory_id=None,
+                    reason=f"Failed gate '{gate_name}': {message}", gates=gates,
+                )
+    memory_id = store.store(content, room=room, category=category, trust=trust)
+    store.log_promotion(session_id, scratch_key, memory_id, reason="auto" if not force else "forced")
+    return PromotionResult(success=True, memory_id=memory_id, reason="Promoted to durable memory", gates=gates)
+
+
+def promote_session_batch(
+    store: SovereignStore,
+    session_id: str,
+    notes: dict[str, dict],
+    room: str = "general",
+    force: bool = False,
+) -> list[PromotionResult]:
+    """Promote all notes from a session scratchpad."""
+    results = []
+    for key, entry in notes.items():
+        content = entry.get("value", str(entry)) if isinstance(entry, dict) else str(entry)
+        written_at = None
+        if isinstance(entry, dict) and "written_at" in entry:
+            try:
+                import datetime
+                written_at = datetime.datetime.strptime(
+                    entry["written_at"], "%Y-%m-%d %H:%M:%S"
+                ).timestamp()
+            except (ValueError, TypeError):
+                pass
+        result = promote(
+            content=str(content), store=store, session_id=session_id,
+            scratch_key=key, room=room, written_at=written_at, force=force,
+        )
+        results.append(result)
+    return results

hermes-sovereign/mempalace/retrieval_enforcer.py

@@ -0,0 +1,310 @@
+"""Retrieval Order Enforcer — L0 through L5 memory hierarchy.
+
+Ensures the agent checks durable memory before falling back to free generation.
+Gracefully degrades if any layer is unavailable (missing files, etc).
+
+Layer order:
+    L0: Identity      (~/.mempalace/identity.txt)
+    L1: Palace rooms   (SovereignStore — SQLite + FTS5 + HRR, zero API calls)
+    L2: Session scratch (~/.hermes/scratchpad/{session_id}.json)
+    L3: Gitea artifacts (API search for issues/PRs)
+    L4: Procedures      (skills directory search)
+    L5: Free generation (only if L0-L4 produced nothing)
+
+Refs: Epic #367, Sub-issue #369, Wiring: #383
+"""
+from __future__ import annotations
+
+import json
+import os
+import re
+from pathlib import Path
+from typing import Optional
+
+# ---------------------------------------------------------------------------
+# Sovereign Store (replaces mempalace CLI subprocess)
+# ---------------------------------------------------------------------------
+try:
+    from .sovereign_store import SovereignStore
+except ImportError:
+    try:
+        from sovereign_store import SovereignStore
+    except ImportError:
+        SovereignStore = None  # type: ignore[misc,assignment]
+
+# ---------------------------------------------------------------------------
+# Constants
+# ---------------------------------------------------------------------------
+
+IDENTITY_PATH = Path.home() / ".mempalace" / "identity.txt"
+SCRATCHPAD_DIR = Path.home() / ".hermes" / "scratchpad"
+SKILLS_DIR = Path.home() / ".hermes" / "skills"
+SOVEREIGN_DB = Path.home() / ".hermes" / "palace" / "sovereign.db"
+
+# Patterns that indicate a recall-style query
+RECALL_PATTERNS = re.compile(
+    r"(?i)\b("
+    r"what did|status of|remember|last time|yesterday|previously|"
+    r"we discussed|we talked|we worked|you said|you mentioned|"
+    r"remind me|what was|what were|how did|when did|"
+    r"earlier today|last session|before this"
+    r")\b"
+)
+
+# Singleton store instance (lazy-init)
+_store: Optional["SovereignStore"] = None
+
+
+def _get_store() -> Optional["SovereignStore"]:
+    """Lazy-init the SovereignStore singleton."""
+    global _store
+    if _store is not None:
+        return _store
+    if SovereignStore is None:
+        return None
+    try:
+        _store = SovereignStore(db_path=str(SOVEREIGN_DB))
+        return _store
+    except Exception:
+        return None
+
+
+# ---------------------------------------------------------------------------
+# L0: Identity
+# ---------------------------------------------------------------------------
+
+def load_identity() -> str:
+    """Read the agent identity file. Returns empty string on failure."""
+    try:
+        if IDENTITY_PATH.exists():
+            text = IDENTITY_PATH.read_text(encoding="utf-8").strip()
+            # Cap at ~200 tokens to keep wake-up lean
+            if len(text.split()) > 200:
+                text = " ".join(text.split()[:200]) + "..."
+            return text
+    except (OSError, PermissionError):
+        pass
+    return ""
+
+
+# ---------------------------------------------------------------------------
+# L1: Palace search (now via SovereignStore — zero subprocess, zero API)
+# ---------------------------------------------------------------------------
+
+def search_palace(query: str, room: Optional[str] = None) -> str:
+    """Search the sovereign memory store for relevant memories.
+
+    Uses SovereignStore (SQLite + FTS5 + HRR) for hybrid keyword + semantic
+    search. No subprocess calls, no ONNX, no API keys.
+
+    Gracefully degrades to empty string if store is unavailable.
+    """
+    store = _get_store()
+    if store is None:
+        return ""
+    try:
+        results = store.search(query, room=room, limit=5, min_trust=0.2)
+        if not results:
+            return ""
+        lines = []
+        for r in results:
+            trust = r.get("trust_score", 0.5)
+            room_name = r.get("room", "general")
+            content = r.get("content", "")
+            lines.append(f"  [{room_name}] (trust:{trust:.2f}) {content}")
+        return "\n".join(lines)
+    except Exception:
+        return ""
+
+
+# ---------------------------------------------------------------------------
+# L2: Session scratchpad
+# ---------------------------------------------------------------------------
+
+def load_scratchpad(session_id: str) -> str:
+    """Load the session scratchpad as formatted text."""
+    try:
+        scratch_file = SCRATCHPAD_DIR / f"{session_id}.json"
+        if scratch_file.exists():
+            data = json.loads(scratch_file.read_text(encoding="utf-8"))
+            if isinstance(data, dict) and data:
+                lines = []
+                for k, v in data.items():
+                    lines.append(f"  {k}: {v}")
+                return "\n".join(lines)
+    except (OSError, json.JSONDecodeError):
+        pass
+    return ""
+
+
+# ---------------------------------------------------------------------------
+# L3: Gitea artifact search
+# ---------------------------------------------------------------------------
+
+def _load_gitea_token() -> str:
+    """Read the Gitea API token."""
+    token_path = Path.home() / ".hermes" / "gitea_token_vps"
+    try:
+        if token_path.exists():
+            return token_path.read_text(encoding="utf-8").strip()
+    except OSError:
+        pass
+    return ""
+
+
+def search_gitea(query: str) -> str:
+    """Search Gitea issues/PRs for context. Returns formatted text or empty string."""
+    token = _load_gitea_token()
+    if not token:
+        return ""
+
+    api_base = "https://forge.alexanderwhitestone.com/api/v1"
+    # Extract key terms for search (first 3 significant words)
+    terms = [w for w in query.split() if len(w) > 3][:3]
+    search_q = " ".join(terms) if terms else query[:50]
+
+    try:
+        import urllib.request
+        import urllib.parse
+
+        url = (
+            f"{api_base}/repos/search?"
+            f"q={urllib.parse.quote(search_q)}&limit=3"
+        )
+        req = urllib.request.Request(url, headers={
+            "Authorization": f"token {token}",
+            "Accept": "application/json",
+        })
+        with urllib.request.urlopen(req, timeout=8) as resp:
+            data = json.loads(resp.read().decode())
+            if data.get("data"):
+                lines = []
+                for repo in data["data"][:3]:
+                    lines.append(f"  {repo['full_name']}: {repo.get('description', 'no desc')}")
+                return "\n".join(lines)
+    except Exception:
+        pass
+    return ""
+
+
+# ---------------------------------------------------------------------------
+# L4: Procedures (skills search)
+# ---------------------------------------------------------------------------
+
+def search_skills(query: str) -> str:
+    """Search skills directory for matching procedures."""
+    try:
+        if not SKILLS_DIR.exists():
+            return ""
+
+        query_lower = query.lower()
+        terms = [w for w in query_lower.split() if len(w) > 3]
+        if not terms:
+            return ""
+
+        matches = []
+        for skill_dir in SKILLS_DIR.iterdir():
+            if not skill_dir.is_dir():
+                continue
+            skill_md = skill_dir / "SKILL.md"
+            if skill_md.exists():
+                try:
+                    content = skill_md.read_text(encoding="utf-8").lower()
+                    if any(t in content for t in terms):
+                        title = skill_dir.name
+                        matches.append(f"  skill: {title}")
+                except OSError:
+                    continue
+
+        if matches:
+            return "\n".join(matches[:5])
+    except OSError:
+        pass
+    return ""
+
+
+# ---------------------------------------------------------------------------
+# Main enforcer
+# ---------------------------------------------------------------------------
+
+def is_recall_query(query: str) -> bool:
+    """Detect whether a query is asking for recalled/historical information."""
+    return bool(RECALL_PATTERNS.search(query))
+
+
+def enforce_retrieval_order(
+    query: str,
+    session_id: Optional[str] = None,
+    skip_if_not_recall: bool = True,
+) -> dict:
+    """Check palace layers before allowing free generation.
+
+    Args:
+        query: The user's query text.
+        session_id: Current session ID for scratchpad access.
+        skip_if_not_recall: If True (default), skip enforcement for
+            non-recall queries and return empty result.
+
+    Returns:
+        dict with keys:
+            retrieved_from: Highest layer that produced results (e.g. 'L1')
+            context: Aggregated context string
+            tokens: Approximate word count of context
+            layers_checked: List of layers that were consulted
+    """
+    result = {
+        "retrieved_from": None,
+        "context": "",
+        "tokens": 0,
+        "layers_checked": [],
+    }
+
+    # Gate: skip for non-recall queries if configured
+    if skip_if_not_recall and not is_recall_query(query):
+        return result
+
+    # L0: Identity (always prepend)
+    identity = load_identity()
+    if identity:
+        result["context"] += f"## Identity\n{identity}\n\n"
+        result["layers_checked"].append("L0")
+
+    # L1: Palace search (SovereignStore — zero API, zero subprocess)
+    palace_results = search_palace(query)
+    if palace_results:
+        result["context"] += f"## Palace Memory\n{palace_results}\n\n"
+        result["retrieved_from"] = "L1"
+        result["layers_checked"].append("L1")
+
+    # L2: Scratchpad
+    if session_id:
+        scratch = load_scratchpad(session_id)
+        if scratch:
+            result["context"] += f"## Session Notes\n{scratch}\n\n"
+            if not result["retrieved_from"]:
+                result["retrieved_from"] = "L2"
+            result["layers_checked"].append("L2")
+
+    # L3: Gitea artifacts (only if still no context from L1/L2)
+    if not result["retrieved_from"]:
+        artifacts = search_gitea(query)
+        if artifacts:
+            result["context"] += f"## Gitea Context\n{artifacts}\n\n"
+            result["retrieved_from"] = "L3"
+            result["layers_checked"].append("L3")
+
+    # L4: Procedures (only if still no context)
+    if not result["retrieved_from"]:
+        procedures = search_skills(query)
+        if procedures:
+            result["context"] += f"## Related Skills\n{procedures}\n\n"
+            result["retrieved_from"] = "L4"
+            result["layers_checked"].append("L4")
+
+    # L5: Free generation (no context found — just mark it)
+    if not result["retrieved_from"]:
+        result["retrieved_from"] = "L5"
+        result["layers_checked"].append("L5")
+
+    result["tokens"] = len(result["context"].split())
+    return result

hermes-sovereign/mempalace/scratchpad.py

@@ -0,0 +1,184 @@
+"""Session Scratchpad — ephemeral key-value notes per session.
+
+Provides fast, JSON-backed scratch storage that lives for a session
+and can be promoted to durable palace memory.
+
+Storage: ~/.hermes/scratchpad/{session_id}.json
+
+Refs: Epic #367, Sub-issue #372
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import subprocess
+import time
+from pathlib import Path
+from typing import Any, Optional
+
+# ---------------------------------------------------------------------------
+# Constants
+# ---------------------------------------------------------------------------
+
+SCRATCHPAD_DIR = Path.home() / ".hermes" / "scratchpad"
+MEMPALACE_BIN = "/Library/Frameworks/Python.framework/Versions/3.12/bin/mempalace"
+
+
+# ---------------------------------------------------------------------------
+# Internal helpers
+# ---------------------------------------------------------------------------
+
+def _scratch_path(session_id: str) -> Path:
+    """Return the JSON file path for a given session."""
+    # Sanitize session_id to prevent path traversal
+    safe_id = "".join(c for c in session_id if c.isalnum() or c in "-_")
+    if not safe_id:
+        safe_id = "unnamed"
+    return SCRATCHPAD_DIR / f"{safe_id}.json"
+
+
+def _load(session_id: str) -> dict:
+    """Load scratchpad data, returning empty dict on failure."""
+    path = _scratch_path(session_id)
+    try:
+        if path.exists():
+            return json.loads(path.read_text(encoding="utf-8"))
+    except (OSError, json.JSONDecodeError):
+        pass
+    return {}
+
+
+def _save(session_id: str, data: dict) -> None:
+    """Persist scratchpad data to disk."""
+    SCRATCHPAD_DIR.mkdir(parents=True, exist_ok=True)
+    path = _scratch_path(session_id)
+    path.write_text(json.dumps(data, indent=2, default=str), encoding="utf-8")
+
+
+# ---------------------------------------------------------------------------
+# Public API
+# ---------------------------------------------------------------------------
+
+def write_scratch(session_id: str, key: str, value: Any) -> None:
+    """Write a note to the session scratchpad.
+
+    Args:
+        session_id: Current session identifier.
+        key: Note key (string).
+        value: Note value (any JSON-serializable type).
+    """
+    data = _load(session_id)
+    data[key] = {
+        "value": value,
+        "written_at": time.strftime("%Y-%m-%d %H:%M:%S"),
+    }
+    _save(session_id, data)
+
+
+def read_scratch(session_id: str, key: Optional[str] = None) -> dict:
+    """Read session scratchpad (all keys or one).
+
+    Args:
+        session_id: Current session identifier.
+        key: Optional specific key. If None, returns all entries.
+
+    Returns:
+        dict — either {key: {value, written_at}} or the full scratchpad.
+    """
+    data = _load(session_id)
+    if key is not None:
+        entry = data.get(key)
+        return {key: entry} if entry else {}
+    return data
+
+
+def delete_scratch(session_id: str, key: str) -> bool:
+    """Remove a single key from the scratchpad.
+
+    Returns True if the key existed and was removed.
+    """
+    data = _load(session_id)
+    if key in data:
+        del data[key]
+        _save(session_id, data)
+        return True
+    return False
+
+
+def list_sessions() -> list[str]:
+    """List all session IDs that have scratchpad files."""
+    try:
+        if SCRATCHPAD_DIR.exists():
+            return [
+                f.stem
+                for f in SCRATCHPAD_DIR.iterdir()
+                if f.suffix == ".json" and f.is_file()
+            ]
+    except OSError:
+        pass
+    return []
+
+
+def promote_to_palace(
+    session_id: str,
+    key: str,
+    room: str = "general",
+    drawer: Optional[str] = None,
+) -> bool:
+    """Move a scratchpad note to durable palace memory.
+
+    Uses the mempalace CLI to store the note in the specified room.
+    Removes the note from the scratchpad after successful promotion.
+
+    Args:
+        session_id: Session containing the note.
+        key: Scratchpad key to promote.
+        room: Palace room name (default: 'general').
+        drawer: Optional drawer name within the room. Defaults to key.
+
+    Returns:
+        True if promotion succeeded, False otherwise.
+    """
+    data = _load(session_id)
+    entry = data.get(key)
+    if not entry:
+        return False
+
+    value = entry.get("value", entry) if isinstance(entry, dict) else entry
+    content = json.dumps(value, default=str) if not isinstance(value, str) else value
+
+    try:
+        bin_path = MEMPALACE_BIN if os.path.exists(MEMPALACE_BIN) else "mempalace"
+        target_drawer = drawer or key
+        result = subprocess.run(
+            [bin_path, "store", room, target_drawer, content],
+            capture_output=True,
+            text=True,
+            timeout=10,
+        )
+        if result.returncode == 0:
+            # Remove from scratchpad after successful promotion
+            del data[key]
+            _save(session_id, data)
+            return True
+    except (FileNotFoundError, subprocess.TimeoutExpired, OSError):
+        # mempalace CLI not available — degrade gracefully
+        pass
+
+    return False
+
+
+def clear_session(session_id: str) -> bool:
+    """Delete the entire scratchpad for a session.
+
+    Returns True if the file existed and was removed.
+    """
+    path = _scratch_path(session_id)
+    try:
+        if path.exists():
+            path.unlink()
+            return True
+    except OSError:
+        pass
+    return False

hermes-sovereign/mempalace/sovereign_store.py

@@ -0,0 +1,474 @@
+"""Sovereign Memory Store — zero-API, zero-dependency durable memory.
+
+Replaces the third-party `mempalace` CLI and its ONNX requirement with a
+self-contained SQLite + FTS5 + HRR (Holographic Reduced Representation)
+store. Every operation is local: no network calls, no API keys, no cloud.
+
+Storage: ~/.hermes/palace/sovereign.db
+
+Capabilities:
+  - Durable fact storage with rooms, categories, and trust scores
+  - Hybrid retrieval: FTS5 keyword search + HRR cosine similarity
+  - Reciprocal Rank Fusion to merge keyword and semantic results
+  - Trust scoring: facts that get retrieved and confirmed gain trust
+  - Graceful numpy degradation: falls back to keyword-only if missing
+
+Refs: Epic #367, MP-3 #370, MP-4 #371
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import math
+import sqlite3
+import struct
+import time
+from pathlib import Path
+from typing import Any, Optional
+
+# ---------------------------------------------------------------------------
+# HRR (Holographic Reduced Representations) — zero-dependency vectors
+# ---------------------------------------------------------------------------
+# Phase-encoded vectors via SHA-256. No ONNX, no embeddings API, no numpy
+# required (but uses numpy when available for speed).
+
+_TWO_PI = 2.0 * math.pi
+_DIM = 512  # Compact dimension — sufficient for memory retrieval
+
+try:
+    import numpy as np
+    _HAS_NUMPY = True
+except ImportError:
+    _HAS_NUMPY = False
+
+
+def _encode_atom_np(word: str, dim: int = _DIM) -> "np.ndarray":
+    """Deterministic phase vector via SHA-256 (numpy path)."""
+    values_per_block = 16
+    blocks_needed = math.ceil(dim / values_per_block)
+    uint16_values: list[int] = []
+    for i in range(blocks_needed):
+        digest = hashlib.sha256(f"{word}:{i}".encode()).digest()
+        uint16_values.extend(struct.unpack("<16H", digest))
+    return np.array(uint16_values[:dim], dtype=np.float64) * (_TWO_PI / 65536.0)
+
+
+def _encode_atom_pure(word: str, dim: int = _DIM) -> list[float]:
+    """Deterministic phase vector via SHA-256 (pure Python fallback)."""
+    values_per_block = 16
+    blocks_needed = math.ceil(dim / values_per_block)
+    uint16_values: list[int] = []
+    for i in range(blocks_needed):
+        digest = hashlib.sha256(f"{word}:{i}".encode()).digest()
+        for j in range(0, 32, 2):
+            uint16_values.append(int.from_bytes(digest[j:j+2], "little"))
+    return [v * (_TWO_PI / 65536.0) for v in uint16_values[:dim]]
+
+
+def encode_text(text: str, dim: int = _DIM):
+    """Encode a text string into an HRR phase vector by bundling word atoms.
+
+    Uses circular mean of per-word phase vectors — the standard HRR
+    superposition operation. Result is a fixed-width vector regardless
+    of input length.
+    """
+    words = text.lower().split()
+    if not words:
+        words = ["<empty>"]
+
+    if _HAS_NUMPY:
+        atoms = [_encode_atom_np(w, dim) for w in words]
+        # Circular mean: average the unit vectors, extract phase
+        unit_sum = sum(np.exp(1j * a) for a in atoms)
+        return np.angle(unit_sum) % _TWO_PI
+    else:
+        # Pure Python circular mean
+        real_sum = [0.0] * dim
+        imag_sum = [0.0] * dim
+        for w in words:
+            atom = _encode_atom_pure(w, dim)
+            for d in range(dim):
+                real_sum[d] += math.cos(atom[d])
+                imag_sum[d] += math.sin(atom[d])
+        return [math.atan2(imag_sum[d], real_sum[d]) % _TWO_PI for d in range(dim)]
+
+
+def cosine_similarity_phase(a, b) -> float:
+    """Cosine similarity between two phase vectors.
+
+    For phase vectors, similarity = mean(cos(a - b)).
+    """
+    if _HAS_NUMPY:
+        return float(np.mean(np.cos(np.array(a) - np.array(b))))
+    else:
+        n = len(a)
+        return sum(math.cos(a[i] - b[i]) for i in range(n)) / n
+
+
+def serialize_vector(vec) -> bytes:
+    """Serialize a vector to bytes for SQLite storage."""
+    if _HAS_NUMPY:
+        return vec.astype(np.float64).tobytes()
+    else:
+        return struct.pack(f"{len(vec)}d", *vec)
+
+
+def deserialize_vector(blob: bytes):
+    """Deserialize bytes back to a vector."""
+    n = len(blob) // 8  # float64 = 8 bytes
+    if _HAS_NUMPY:
+        return np.frombuffer(blob, dtype=np.float64)
+    else:
+        return list(struct.unpack(f"{n}d", blob))
+
+
+# ---------------------------------------------------------------------------
+# SQLite Schema
+# ---------------------------------------------------------------------------
+
+_SCHEMA = """
+CREATE TABLE IF NOT EXISTS memories (
+    memory_id       INTEGER PRIMARY KEY AUTOINCREMENT,
+    content         TEXT NOT NULL,
+    room            TEXT DEFAULT 'general',
+    category        TEXT DEFAULT '',
+    trust_score     REAL DEFAULT 0.5,
+    retrieval_count INTEGER DEFAULT 0,
+    created_at      REAL NOT NULL,
+    updated_at      REAL NOT NULL,
+    hrr_vector      BLOB
+);
+
+CREATE INDEX IF NOT EXISTS idx_memories_room ON memories(room);
+CREATE INDEX IF NOT EXISTS idx_memories_trust ON memories(trust_score DESC);
+
+-- FTS5 for fast keyword search
+CREATE VIRTUAL TABLE IF NOT EXISTS memories_fts USING fts5(
+    content, room, category,
+    content=memories, content_rowid=memory_id,
+    tokenize='porter unicode61'
+);
+
+-- Sync triggers
+CREATE TRIGGER IF NOT EXISTS memories_ai AFTER INSERT ON memories BEGIN
+    INSERT INTO memories_fts(rowid, content, room, category)
+        VALUES (new.memory_id, new.content, new.room, new.category);
+END;
+
+CREATE TRIGGER IF NOT EXISTS memories_ad AFTER DELETE ON memories BEGIN
+    INSERT INTO memories_fts(memories_fts, rowid, content, room, category)
+        VALUES ('delete', old.memory_id, old.content, old.room, old.category);
+END;
+
+CREATE TRIGGER IF NOT EXISTS memories_au AFTER UPDATE ON memories BEGIN
+    INSERT INTO memories_fts(memories_fts, rowid, content, room, category)
+        VALUES ('delete', old.memory_id, old.content, old.room, old.category);
+    INSERT INTO memories_fts(rowid, content, room, category)
+        VALUES (new.memory_id, new.content, new.room, new.category);
+END;
+
+-- Promotion log: tracks what moved from scratchpad to durable memory
+CREATE TABLE IF NOT EXISTS promotion_log (
+    log_id      INTEGER PRIMARY KEY AUTOINCREMENT,
+    session_id  TEXT NOT NULL,
+    scratch_key TEXT NOT NULL,
+    memory_id   INTEGER REFERENCES memories(memory_id),
+    promoted_at REAL NOT NULL,
+    reason      TEXT DEFAULT ''
+);
+"""
+
+
+# ---------------------------------------------------------------------------
+# SovereignStore
+# ---------------------------------------------------------------------------
+
+class SovereignStore:
+    """Zero-API durable memory store.
+
+    All operations are local SQLite. No network calls. No API keys.
+    HRR vectors provide semantic similarity without embedding models.
+    FTS5 provides fast keyword search. RRF merges both rankings.
+    """
+
+    def __init__(self, db_path: Optional[str] = None):
+        if db_path is None:
+            db_path = str(Path.home() / ".hermes" / "palace" / "sovereign.db")
+        self._db_path = db_path
+        Path(db_path).parent.mkdir(parents=True, exist_ok=True)
+        self._conn = sqlite3.connect(db_path)
+        self._conn.row_factory = sqlite3.Row
+        self._conn.executescript(_SCHEMA)
+
+    def close(self):
+        self._conn.close()
+
+    # ------------------------------------------------------------------
+    # Store
+    # ------------------------------------------------------------------
+
+    def store(
+        self,
+        content: str,
+        room: str = "general",
+        category: str = "",
+        trust: float = 0.5,
+    ) -> int:
+        """Store a fact in durable memory. Returns the memory_id."""
+        now = time.time()
+        vec = encode_text(content)
+        blob = serialize_vector(vec)
+        cur = self._conn.execute(
+            """INSERT INTO memories (content, room, category, trust_score,
+               created_at, updated_at, hrr_vector)
+               VALUES (?, ?, ?, ?, ?, ?, ?)""",
+            (content, room, category, trust, now, now, blob),
+        )
+        self._conn.commit()
+        return cur.lastrowid
+
+    def store_batch(self, items: list[dict]) -> list[int]:
+        """Store multiple facts. Each item: {content, room?, category?, trust?}."""
+        ids = []
+        now = time.time()
+        for item in items:
+            content = item["content"]
+            vec = encode_text(content)
+            blob = serialize_vector(vec)
+            cur = self._conn.execute(
+                """INSERT INTO memories (content, room, category, trust_score,
+                   created_at, updated_at, hrr_vector)
+                   VALUES (?, ?, ?, ?, ?, ?, ?)""",
+                (
+                    content,
+                    item.get("room", "general"),
+                    item.get("category", ""),
+                    item.get("trust", 0.5),
+                    now, now, blob,
+                ),
+            )
+            ids.append(cur.lastrowid)
+        self._conn.commit()
+        return ids
+
+    # ------------------------------------------------------------------
+    # Search — hybrid FTS5 + HRR with Reciprocal Rank Fusion
+    # ------------------------------------------------------------------
+
+    def search(
+        self,
+        query: str,
+        room: Optional[str] = None,
+        limit: int = 10,
+        min_trust: float = 0.0,
+        fts_weight: float = 0.5,
+        hrr_weight: float = 0.5,
+    ) -> list[dict]:
+        """Hybrid search: FTS5 keywords + HRR semantic similarity.
+
+        Uses Reciprocal Rank Fusion (RRF) to merge both rankings.
+        Returns list of dicts with content, room, score, trust_score.
+        """
+        k_rrf = 60  # Standard RRF constant
+
+        # Stage 1: FTS5 candidates
+        fts_results = self._fts_search(query, room, min_trust, limit * 3)
+
+        # Stage 2: HRR candidates (scan top N by trust)
+        hrr_results = self._hrr_search(query, room, min_trust, limit * 3)
+
+        # Stage 3: RRF fusion
+        scores: dict[int, float] = {}
+        meta: dict[int, dict] = {}
+
+        for rank, row in enumerate(fts_results):
+            mid = row["memory_id"]
+            scores[mid] = scores.get(mid, 0) + fts_weight / (k_rrf + rank + 1)
+            meta[mid] = dict(row)
+
+        for rank, row in enumerate(hrr_results):
+            mid = row["memory_id"]
+            scores[mid] = scores.get(mid, 0) + hrr_weight / (k_rrf + rank + 1)
+            if mid not in meta:
+                meta[mid] = dict(row)
+
+        # Sort by fused score
+        ranked = sorted(scores.items(), key=lambda x: x[1], reverse=True)[:limit]
+
+        results = []
+        for mid, score in ranked:
+            m = meta[mid]
+            # Bump retrieval count
+            self._conn.execute(
+                "UPDATE memories SET retrieval_count = retrieval_count + 1 WHERE memory_id = ?",
+                (mid,),
+            )
+            results.append({
+                "memory_id": mid,
+                "content": m["content"],
+                "room": m["room"],
+                "category": m.get("category", ""),
+                "trust_score": m["trust_score"],
+                "score": round(score, 6),
+            })
+
+        if results:
+            self._conn.commit()
+        return results
+
+    def _fts_search(
+        self, query: str, room: Optional[str], min_trust: float, limit: int
+    ) -> list[dict]:
+        """FTS5 full-text search."""
+        try:
+            if room:
+                rows = self._conn.execute(
+                    """SELECT m.memory_id, m.content, m.room, m.category,
+                              m.trust_score, m.retrieval_count
+                       FROM memories_fts f
+                       JOIN memories m ON f.rowid = m.memory_id
+                       WHERE memories_fts MATCH ? AND m.room = ?
+                             AND m.trust_score >= ?
+                       ORDER BY rank LIMIT ?""",
+                    (query, room, min_trust, limit),
+                ).fetchall()
+            else:
+                rows = self._conn.execute(
+                    """SELECT m.memory_id, m.content, m.room, m.category,
+                              m.trust_score, m.retrieval_count
+                       FROM memories_fts f
+                       JOIN memories m ON f.rowid = m.memory_id
+                       WHERE memories_fts MATCH ?
+                             AND m.trust_score >= ?
+                       ORDER BY rank LIMIT ?""",
+                    (query, min_trust, limit),
+                ).fetchall()
+            return [dict(r) for r in rows]
+        except sqlite3.OperationalError:
+            # Bad FTS query syntax — degrade gracefully
+            return []
+
+    def _hrr_search(
+        self, query: str, room: Optional[str], min_trust: float, limit: int
+    ) -> list[dict]:
+        """HRR cosine similarity search (brute-force scan, fast for <100K facts)."""
+        query_vec = encode_text(query)
+
+        if room:
+            rows = self._conn.execute(
+                """SELECT memory_id, content, room, category, trust_score,
+                          retrieval_count, hrr_vector
+                   FROM memories
+                   WHERE room = ? AND trust_score >= ? AND hrr_vector IS NOT NULL""",
+                (room, min_trust),
+            ).fetchall()
+        else:
+            rows = self._conn.execute(
+                """SELECT memory_id, content, room, category, trust_score,
+                          retrieval_count, hrr_vector
+                   FROM memories
+                   WHERE trust_score >= ? AND hrr_vector IS NOT NULL""",
+                (min_trust,),
+            ).fetchall()
+
+        scored = []
+        for r in rows:
+            stored_vec = deserialize_vector(r["hrr_vector"])
+            sim = cosine_similarity_phase(query_vec, stored_vec)
+            scored.append((sim, dict(r)))
+
+        scored.sort(key=lambda x: x[0], reverse=True)
+        return [item[1] for item in scored[:limit]]
+
+    # ------------------------------------------------------------------
+    # Trust management
+    # ------------------------------------------------------------------
+
+    def boost_trust(self, memory_id: int, delta: float = 0.05) -> None:
+        """Increase trust score when a memory proves useful."""
+        self._conn.execute(
+            """UPDATE memories SET trust_score = MIN(1.0, trust_score + ?),
+               updated_at = ? WHERE memory_id = ?""",
+            (delta, time.time(), memory_id),
+        )
+        self._conn.commit()
+
+    def decay_trust(self, memory_id: int, delta: float = 0.02) -> None:
+        """Decrease trust score when a memory is contradicted."""
+        self._conn.execute(
+            """UPDATE memories SET trust_score = MAX(0.0, trust_score - ?),
+               updated_at = ? WHERE memory_id = ?""",
+            (delta, time.time(), memory_id),
+        )
+        self._conn.commit()
+
+    # ------------------------------------------------------------------
+    # Room operations
+    # ------------------------------------------------------------------
+
+    def list_rooms(self) -> list[dict]:
+        """List all rooms with fact counts."""
+        rows = self._conn.execute(
+            """SELECT room, COUNT(*) as count,
+                      AVG(trust_score) as avg_trust
+               FROM memories GROUP BY room ORDER BY count DESC"""
+        ).fetchall()
+        return [dict(r) for r in rows]
+
+    def room_contents(self, room: str, limit: int = 50) -> list[dict]:
+        """Get all facts in a room, ordered by trust."""
+        rows = self._conn.execute(
+            """SELECT memory_id, content, category, trust_score,
+                      retrieval_count, created_at
+               FROM memories WHERE room = ?
+               ORDER BY trust_score DESC, created_at DESC LIMIT ?""",
+            (room, limit),
+        ).fetchall()
+        return [dict(r) for r in rows]
+
+    # ------------------------------------------------------------------
+    # Stats
+    # ------------------------------------------------------------------
+
+    def stats(self) -> dict:
+        """Return store statistics."""
+        row = self._conn.execute(
+            """SELECT COUNT(*) as total,
+                      AVG(trust_score) as avg_trust,
+                      SUM(retrieval_count) as total_retrievals,
+                      COUNT(DISTINCT room) as room_count
+               FROM memories"""
+        ).fetchone()
+        return dict(row)
+
+    # ------------------------------------------------------------------
+    # Promotion support (scratchpad → durable)
+    # ------------------------------------------------------------------
+
+    def log_promotion(
+        self,
+        session_id: str,
+        scratch_key: str,
+        memory_id: int,
+        reason: str = "",
+    ) -> None:
+        """Record a scratchpad-to-palace promotion in the audit log."""
+        self._conn.execute(
+            """INSERT INTO promotion_log
+               (session_id, scratch_key, memory_id, promoted_at, reason)
+               VALUES (?, ?, ?, ?, ?)""",
+            (session_id, scratch_key, memory_id, time.time(), reason),
+        )
+        self._conn.commit()
+
+    def recent_promotions(self, limit: int = 20) -> list[dict]:
+        """Get recent promotion log entries."""
+        rows = self._conn.execute(
+            """SELECT p.*, m.content, m.room
+               FROM promotion_log p
+               LEFT JOIN memories m ON p.memory_id = m.memory_id
+               ORDER BY p.promoted_at DESC LIMIT ?""",
+            (limit,),
+        ).fetchall()
+        return [dict(r) for r in rows]

hermes-sovereign/mempalace/tests/test_mempalace.py

@@ -0,0 +1,180 @@
+"""Tests for the mempalace skill.
+
+Validates PalaceRoom, Mempalace class, factory constructors,
+and the analyse_issues entry-point.
+
+Refs: Epic #367, Sub-issue #368
+"""
+
+from __future__ import annotations
+
+import json
+import sys
+import os
+import time
+
+import pytest
+
+# Ensure the package is importable from the repo layout
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", ".."))
+
+from mempalace.mempalace import Mempalace, PalaceRoom, analyse_issues
+
+
+# ── PalaceRoom unit tests ─────────────────────────────────────────────────
+
+class TestPalaceRoom:
+    def test_store_and_retrieve(self):
+        room = PalaceRoom(name="test", label="Test Room")
+        room.store("key1", 42)
+        assert room.retrieve("key1") == 42
+
+    def test_retrieve_default(self):
+        room = PalaceRoom(name="test", label="Test Room")
+        assert room.retrieve("missing") is None
+        assert room.retrieve("missing", "fallback") == "fallback"
+
+    def test_summary_format(self):
+        room = PalaceRoom(name="test", label="Test Room")
+        room.store("repos", 5)
+        summary = room.summary()
+        assert "## Test Room" in summary
+        assert "repos: 5" in summary
+
+    def test_contents_default_factory_isolation(self):
+        """Each room gets its own dict — no shared mutable default."""
+        r1 = PalaceRoom(name="a", label="A")
+        r2 = PalaceRoom(name="b", label="B")
+        r1.store("x", 1)
+        assert r2.retrieve("x") is None
+
+    def test_entered_at_is_recent(self):
+        before = time.time()
+        room = PalaceRoom(name="t", label="T")
+        after = time.time()
+        assert before <= room.entered_at <= after
+
+
+# ── Mempalace core tests ──────────────────────────────────────────────────
+
+class TestMempalace:
+    def test_add_and_enter_room(self):
+        p = Mempalace(domain="test")
+        p.add_room("r1", "Room 1")
+        room = p.enter("r1")
+        assert room.name == "r1"
+
+    def test_enter_nonexistent_room_raises(self):
+        p = Mempalace()
+        with pytest.raises(KeyError, match="No room"):
+            p.enter("ghost")
+
+    def test_store_without_enter_raises(self):
+        p = Mempalace()
+        p.add_room("r", "R")
+        with pytest.raises(RuntimeError, match="Enter a room"):
+            p.store("k", "v")
+
+    def test_store_and_retrieve_via_palace(self):
+        p = Mempalace()
+        p.add_room("r", "R")
+        p.enter("r")
+        p.store("count", 10)
+        assert p.retrieve("r", "count") == 10
+
+    def test_retrieve_missing_room_returns_default(self):
+        p = Mempalace()
+        assert p.retrieve("nope", "key") is None
+        assert p.retrieve("nope", "key", 99) == 99
+
+    def test_render_includes_domain(self):
+        p = Mempalace(domain="audit")
+        p.add_room("r", "Room")
+        p.enter("r")
+        p.store("item", "value")
+        output = p.render()
+        assert "audit" in output
+        assert "Room" in output
+
+    def test_to_dict_structure(self):
+        p = Mempalace(domain="test")
+        p.add_room("r", "R")
+        p.enter("r")
+        p.store("a", 1)
+        d = p.to_dict()
+        assert d["domain"] == "test"
+        assert "elapsed_seconds" in d
+        assert d["rooms"]["r"] == {"a": 1}
+
+    def test_to_json_is_valid(self):
+        p = Mempalace(domain="j")
+        p.add_room("x", "X")
+        p.enter("x")
+        p.store("v", [1, 2, 3])
+        parsed = json.loads(p.to_json())
+        assert parsed["rooms"]["x"]["v"] == [1, 2, 3]
+
+
+# ── Factory constructor tests ─────────────────────────────────────────────
+
+class TestFactories:
+    def test_for_issue_analysis_rooms(self):
+        p = Mempalace.for_issue_analysis()
+        assert p.domain == "issue_analysis"
+        for key in ("repo_architecture", "assignment_status",
+                    "triage_priority", "resolution_patterns"):
+            p.enter(key)  # should not raise
+
+    def test_for_health_check_rooms(self):
+        p = Mempalace.for_health_check()
+        assert p.domain == "health_check"
+        for key in ("service_topology", "failure_signals", "recovery_history"):
+            p.enter(key)
+
+    def test_for_code_review_rooms(self):
+        p = Mempalace.for_code_review()
+        assert p.domain == "code_review"
+        for key in ("change_scope", "risk_surface",
+                    "test_coverage", "reviewer_context"):
+            p.enter(key)
+
+
+# ── analyse_issues entry-point tests ──────────────────────────────────────
+
+class TestAnalyseIssues:
+    SAMPLE_DATA = [
+        {"repo": "the-nexus", "open_issues": 40, "assigned": 30, "unassigned": 10},
+        {"repo": "timmy-home", "open_issues": 30, "assigned": 25, "unassigned": 5},
+        {"repo": "hermes-agent", "open_issues": 20, "assigned": 15, "unassigned": 5},
+        {"repo": "empty-repo", "open_issues": 0, "assigned": 0, "unassigned": 0},
+    ]
+
+    def test_returns_string(self):
+        result = analyse_issues(self.SAMPLE_DATA)
+        assert isinstance(result, str)
+        assert len(result) > 0
+
+    def test_contains_room_headers(self):
+        result = analyse_issues(self.SAMPLE_DATA)
+        assert "Repository Architecture" in result
+        assert "Assignment Status" in result
+
+    def test_coverage_below_target(self):
+        result = analyse_issues(self.SAMPLE_DATA, target_assignee_rate=0.90)
+        assert "BELOW TARGET" in result
+
+    def test_coverage_meets_target(self):
+        good_data = [
+            {"repo": "a", "open_issues": 10, "assigned": 10, "unassigned": 0},
+        ]
+        result = analyse_issues(good_data, target_assignee_rate=0.80)
+        assert "OK" in result
+
+    def test_empty_repos_list(self):
+        result = analyse_issues([])
+        assert isinstance(result, str)
+
+    def test_single_repo(self):
+        data = [{"repo": "solo", "open_issues": 5, "assigned": 3, "unassigned": 2}]
+        result = analyse_issues(data)
+        assert "solo" in result or "issue_analysis" in result

hermes-sovereign/mempalace/tests/test_retrieval_enforcer.py

@@ -0,0 +1,143 @@
+"""Tests for retrieval_enforcer.py.
+
+Refs: Epic #367, Sub-issue #369
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import sys
+import tempfile
+from pathlib import Path
+from unittest.mock import patch, MagicMock
+
+import pytest
+
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", ".."))
+
+from mempalace.retrieval_enforcer import (
+    is_recall_query,
+    load_identity,
+    load_scratchpad,
+    enforce_retrieval_order,
+    search_skills,
+    RECALL_PATTERNS,
+)
+
+
+class TestRecallDetection:
+    """Test the recall-query pattern matcher."""
+
+    @pytest.mark.parametrize("query", [
+        "what did we work on yesterday",
+        "status of the mempalace integration",
+        "remember the fleet audit results",
+        "last time we deployed the nexus",
+        "previously you mentioned a CI fix",
+        "we discussed the sovereign deployment",
+    ])
+    def test_recall_queries_detected(self, query):
+        assert is_recall_query(query) is True
+
+    @pytest.mark.parametrize("query", [
+        "create a new file called test.py",
+        "run the test suite",
+        "deploy to production",
+        "write a function that sums numbers",
+        "install the package",
+    ])
+    def test_non_recall_queries_skipped(self, query):
+        assert is_recall_query(query) is False
+
+
+class TestLoadIdentity:
+    def test_loads_existing_identity(self, tmp_path):
+        identity_file = tmp_path / "identity.txt"
+        identity_file.write_text("I am Timmy. A sovereign AI.")
+        with patch("mempalace.retrieval_enforcer.IDENTITY_PATH", identity_file):
+            result = load_identity()
+            assert "Timmy" in result
+
+    def test_returns_empty_on_missing_file(self, tmp_path):
+        identity_file = tmp_path / "nonexistent.txt"
+        with patch("mempalace.retrieval_enforcer.IDENTITY_PATH", identity_file):
+            result = load_identity()
+            assert result == ""
+
+    def test_truncates_long_identity(self, tmp_path):
+        identity_file = tmp_path / "identity.txt"
+        identity_file.write_text(" ".join(["word"] * 300))
+        with patch("mempalace.retrieval_enforcer.IDENTITY_PATH", identity_file):
+            result = load_identity()
+            assert result.endswith("...")
+            assert len(result.split()) <= 201  # 200 words + "..."
+
+
+class TestLoadScratchpad:
+    def test_loads_valid_scratchpad(self, tmp_path):
+        scratch_file = tmp_path / "session123.json"
+        scratch_file.write_text(json.dumps({"note": "test value", "key2": 42}))
+        with patch("mempalace.retrieval_enforcer.SCRATCHPAD_DIR", tmp_path):
+            result = load_scratchpad("session123")
+            assert "note: test value" in result
+            assert "key2: 42" in result
+
+    def test_returns_empty_on_missing_file(self, tmp_path):
+        with patch("mempalace.retrieval_enforcer.SCRATCHPAD_DIR", tmp_path):
+            result = load_scratchpad("nonexistent")
+            assert result == ""
+
+    def test_returns_empty_on_invalid_json(self, tmp_path):
+        scratch_file = tmp_path / "bad.json"
+        scratch_file.write_text("not valid json{{{")
+        with patch("mempalace.retrieval_enforcer.SCRATCHPAD_DIR", tmp_path):
+            result = load_scratchpad("bad")
+            assert result == ""
+
+
+class TestEnforceRetrievalOrder:
+    def test_skips_non_recall_query(self):
+        result = enforce_retrieval_order("create a new file")
+        assert result["retrieved_from"] is None
+        assert result["tokens"] == 0
+
+    def test_runs_for_recall_query(self, tmp_path):
+        identity_file = tmp_path / "identity.txt"
+        identity_file.write_text("I am Timmy.")
+        with patch("mempalace.retrieval_enforcer.IDENTITY_PATH", identity_file), \
+             patch("mempalace.retrieval_enforcer.search_palace", return_value=""), \
+             patch("mempalace.retrieval_enforcer.search_gitea", return_value=""), \
+             patch("mempalace.retrieval_enforcer.search_skills", return_value=""):
+            result = enforce_retrieval_order("what did we work on yesterday")
+            assert "Identity" in result["context"]
+            assert "L0" in result["layers_checked"]
+
+    def test_palace_hit_sets_l1(self, tmp_path):
+        identity_file = tmp_path / "identity.txt"
+        identity_file.write_text("I am Timmy.")
+        with patch("mempalace.retrieval_enforcer.IDENTITY_PATH", identity_file), \
+             patch("mempalace.retrieval_enforcer.search_palace", return_value="Found: fleet audit results"), \
+             patch("mempalace.retrieval_enforcer.search_gitea", return_value=""):
+            result = enforce_retrieval_order("what did we discuss yesterday")
+            assert result["retrieved_from"] == "L1"
+            assert "Palace Memory" in result["context"]
+
+    def test_falls_through_to_l5(self, tmp_path):
+        identity_file = tmp_path / "nonexistent.txt"
+        with patch("mempalace.retrieval_enforcer.IDENTITY_PATH", identity_file), \
+             patch("mempalace.retrieval_enforcer.search_palace", return_value=""), \
+             patch("mempalace.retrieval_enforcer.search_gitea", return_value=""), \
+             patch("mempalace.retrieval_enforcer.search_skills", return_value=""):
+            result = enforce_retrieval_order("remember the old deployment", skip_if_not_recall=True)
+            assert result["retrieved_from"] == "L5"
+
+    def test_force_mode_skips_recall_check(self, tmp_path):
+        identity_file = tmp_path / "identity.txt"
+        identity_file.write_text("I am Timmy.")
+        with patch("mempalace.retrieval_enforcer.IDENTITY_PATH", identity_file), \
+             patch("mempalace.retrieval_enforcer.search_palace", return_value=""), \
+             patch("mempalace.retrieval_enforcer.search_gitea", return_value=""), \
+             patch("mempalace.retrieval_enforcer.search_skills", return_value=""):
+            result = enforce_retrieval_order("deploy now", skip_if_not_recall=False)
+            assert "Identity" in result["context"]

hermes-sovereign/mempalace/tests/test_scratchpad.py

@@ -0,0 +1,108 @@
+"""Tests for scratchpad.py.
+
+Refs: Epic #367, Sub-issue #372
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import sys
+from pathlib import Path
+from unittest.mock import patch
+
+import pytest
+
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", ".."))
+
+from mempalace.scratchpad import (
+    write_scratch,
+    read_scratch,
+    delete_scratch,
+    list_sessions,
+    clear_session,
+    _scratch_path,
+)
+
+
+@pytest.fixture
+def scratch_dir(tmp_path):
+    """Provide a temporary scratchpad directory."""
+    with patch("mempalace.scratchpad.SCRATCHPAD_DIR", tmp_path):
+        yield tmp_path
+
+
+class TestScratchPath:
+    def test_sanitizes_session_id(self):
+        path = _scratch_path("safe-id_123")
+        assert "safe-id_123.json" in str(path)
+
+    def test_strips_dangerous_chars(self):
+        path = _scratch_path("../../etc/passwd")
+        assert ".." not in path.name
+        assert "/" not in path.name
+        # Dots are stripped, so only alphanumeric chars remain
+        assert path.name == "etcpasswd.json"
+
+
+class TestWriteAndRead:
+    def test_write_then_read(self, scratch_dir):
+        write_scratch("sess1", "note", "hello world")
+        result = read_scratch("sess1", "note")
+        assert "note" in result
+        assert result["note"]["value"] == "hello world"
+
+    def test_read_all_keys(self, scratch_dir):
+        write_scratch("sess1", "a", 1)
+        write_scratch("sess1", "b", 2)
+        result = read_scratch("sess1")
+        assert "a" in result
+        assert "b" in result
+
+    def test_read_missing_key(self, scratch_dir):
+        write_scratch("sess1", "exists", "yes")
+        result = read_scratch("sess1", "missing")
+        assert result == {}
+
+    def test_read_missing_session(self, scratch_dir):
+        result = read_scratch("nonexistent")
+        assert result == {}
+
+    def test_overwrite_key(self, scratch_dir):
+        write_scratch("sess1", "key", "v1")
+        write_scratch("sess1", "key", "v2")
+        result = read_scratch("sess1", "key")
+        assert result["key"]["value"] == "v2"
+
+
+class TestDelete:
+    def test_delete_existing_key(self, scratch_dir):
+        write_scratch("sess1", "key", "val")
+        assert delete_scratch("sess1", "key") is True
+        assert read_scratch("sess1", "key") == {}
+
+    def test_delete_missing_key(self, scratch_dir):
+        write_scratch("sess1", "other", "val")
+        assert delete_scratch("sess1", "missing") is False
+
+
+class TestListSessions:
+    def test_lists_sessions(self, scratch_dir):
+        write_scratch("alpha", "k", "v")
+        write_scratch("beta", "k", "v")
+        sessions = list_sessions()
+        assert "alpha" in sessions
+        assert "beta" in sessions
+
+    def test_empty_directory(self, scratch_dir):
+        assert list_sessions() == []
+
+
+class TestClearSession:
+    def test_clears_existing(self, scratch_dir):
+        write_scratch("sess1", "k", "v")
+        assert clear_session("sess1") is True
+        assert read_scratch("sess1") == {}
+
+    def test_clear_nonexistent(self, scratch_dir):
+        assert clear_session("ghost") is False

hermes-sovereign/mempalace/tests/test_sovereign_store.py

@@ -0,0 +1,255 @@
+"""Tests for the Sovereign Memory Store and Promotion system.
+
+Zero-API, zero-network — everything runs against an in-memory SQLite DB.
+"""
+
+import os
+import sys
+import tempfile
+import time
+import unittest
+
+# Allow imports from parent package
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
+
+from sovereign_store import (
+    SovereignStore,
+    encode_text,
+    cosine_similarity_phase,
+    serialize_vector,
+    deserialize_vector,
+)
+from promotion import (
+    evaluate_for_promotion,
+    promote,
+    promote_session_batch,
+)
+
+
+class TestHRRVectors(unittest.TestCase):
+    """Test the HRR encoding and similarity functions."""
+
+    def test_deterministic_encoding(self):
+        """Same text always produces the same vector."""
+        v1 = encode_text("hello world")
+        v2 = encode_text("hello world")
+        self.assertAlmostEqual(cosine_similarity_phase(v1, v2), 1.0, places=5)
+
+    def test_similar_texts_higher_similarity(self):
+        """Related texts should be more similar than unrelated ones."""
+        v_agent = encode_text("agent memory palace retrieval")
+        v_similar = encode_text("agent recall memory search")
+        v_unrelated = encode_text("banana strawberry fruit smoothie")
+        sim_related = cosine_similarity_phase(v_agent, v_similar)
+        sim_unrelated = cosine_similarity_phase(v_agent, v_unrelated)
+        self.assertGreater(sim_related, sim_unrelated)
+
+    def test_serialize_roundtrip(self):
+        """Vectors survive serialization to/from bytes."""
+        vec = encode_text("test serialization")
+        blob = serialize_vector(vec)
+        restored = deserialize_vector(blob)
+        sim = cosine_similarity_phase(vec, restored)
+        self.assertAlmostEqual(sim, 1.0, places=5)
+
+    def test_empty_text(self):
+        """Empty text gets a fallback encoding."""
+        vec = encode_text("")
+        self.assertEqual(len(vec) if hasattr(vec, '__len__') else len(list(vec)), 512)
+
+
+class TestSovereignStore(unittest.TestCase):
+    """Test the SQLite-backed sovereign store."""
+
+    def setUp(self):
+        self.db_path = os.path.join(tempfile.mkdtemp(), "test.db")
+        self.store = SovereignStore(db_path=self.db_path)
+
+    def tearDown(self):
+        self.store.close()
+        if os.path.exists(self.db_path):
+            os.remove(self.db_path)
+
+    def test_store_and_retrieve(self):
+        """Store a fact and find it via search."""
+        mid = self.store.store("Timmy is a sovereign AI agent on Hermes VPS", room="identity")
+        results = self.store.search("sovereign agent", room="identity")
+        self.assertTrue(any(r["memory_id"] == mid for r in results))
+
+    def test_fts_search(self):
+        """FTS5 keyword search works."""
+        self.store.store("The beacon game uses paperclips mechanics", room="projects")
+        self.store.store("Fleet agents handle delegation and dispatch", room="fleet")
+        results = self.store.search("paperclips")
+        self.assertTrue(len(results) > 0)
+        self.assertIn("paperclips", results[0]["content"].lower())
+
+    def test_hrr_search_semantic(self):
+        """HRR similarity finds related content even without exact keywords."""
+        self.store.store("Memory palace rooms organize facts spatially", room="memory")
+        self.store.store("Pizza delivery service runs on weekends", room="unrelated")
+        results = self.store.search("organize knowledge rooms", room="memory")
+        self.assertTrue(len(results) > 0)
+        self.assertIn("palace", results[0]["content"].lower())
+
+    def test_room_filtering(self):
+        """Room filter restricts search scope."""
+        self.store.store("Hermes harness manages tool calls", room="infrastructure")
+        self.store.store("Hermes mythology Greek god", room="lore")
+        results = self.store.search("Hermes", room="infrastructure")
+        self.assertTrue(all(r["room"] == "infrastructure" for r in results))
+
+    def test_trust_boost(self):
+        """Trust score increases when boosted."""
+        mid = self.store.store("fact", trust=0.5)
+        self.store.boost_trust(mid, delta=0.1)
+        results = self.store.room_contents("general")
+        fact = next(r for r in results if r["memory_id"] == mid)
+        self.assertAlmostEqual(fact["trust_score"], 0.6, places=2)
+
+    def test_trust_decay(self):
+        """Trust score decreases when decayed."""
+        mid = self.store.store("questionable fact", trust=0.5)
+        self.store.decay_trust(mid, delta=0.2)
+        results = self.store.room_contents("general")
+        fact = next(r for r in results if r["memory_id"] == mid)
+        self.assertAlmostEqual(fact["trust_score"], 0.3, places=2)
+
+    def test_batch_store(self):
+        """Batch store works."""
+        ids = self.store.store_batch([
+            {"content": "fact one", "room": "test"},
+            {"content": "fact two", "room": "test"},
+            {"content": "fact three", "room": "test"},
+        ])
+        self.assertEqual(len(ids), 3)
+        rooms = self.store.list_rooms()
+        test_room = next(r for r in rooms if r["room"] == "test")
+        self.assertEqual(test_room["count"], 3)
+
+    def test_stats(self):
+        """Stats returns correct counts."""
+        self.store.store("a fact", room="r1")
+        self.store.store("another fact", room="r2")
+        s = self.store.stats()
+        self.assertEqual(s["total"], 2)
+        self.assertEqual(s["room_count"], 2)
+
+    def test_retrieval_count_increments(self):
+        """Retrieval count goes up when a fact is found via search."""
+        self.store.store("unique searchable content xyz123", room="test")
+        self.store.search("xyz123")
+        results = self.store.room_contents("test")
+        self.assertTrue(any(r["retrieval_count"] > 0 for r in results))
+
+
+class TestPromotion(unittest.TestCase):
+    """Test the quality-gated promotion system."""
+
+    def setUp(self):
+        self.db_path = os.path.join(tempfile.mkdtemp(), "promo_test.db")
+        self.store = SovereignStore(db_path=self.db_path)
+
+    def tearDown(self):
+        self.store.close()
+
+    def test_successful_promotion(self):
+        """Good content passes all gates."""
+        result = promote(
+            content="Timmy runs on the Hermes VPS at 143.198.27.163 with local Ollama inference",
+            store=self.store,
+            session_id="test-session-001",
+            scratch_key="vps_info",
+            room="infrastructure",
+        )
+        self.assertTrue(result.success)
+        self.assertIsNotNone(result.memory_id)
+
+    def test_reject_too_short(self):
+        """Short fragments get rejected."""
+        result = promote(
+            content="yes",
+            store=self.store,
+            session_id="test",
+            scratch_key="short",
+        )
+        self.assertFalse(result.success)
+        self.assertIn("Too short", result.reason)
+
+    def test_reject_duplicate(self):
+        """Duplicate content gets rejected."""
+        self.store.store("SOUL.md is the canonical identity document for Timmy", room="identity")
+        result = promote(
+            content="SOUL.md is the canonical identity document for Timmy",
+            store=self.store,
+            session_id="test",
+            scratch_key="soul",
+            room="identity",
+        )
+        self.assertFalse(result.success)
+        self.assertIn("uplicate", result.reason)
+
+    def test_reject_stale(self):
+        """Old notes get flagged as stale."""
+        old_time = time.time() - (86400 * 10)
+        result = promote(
+            content="This is a note from long ago about something important",
+            store=self.store,
+            session_id="test",
+            scratch_key="old",
+            written_at=old_time,
+        )
+        self.assertFalse(result.success)
+        self.assertIn("Stale", result.reason)
+
+    def test_force_bypasses_gates(self):
+        """Force flag overrides quality gates."""
+        result = promote(
+            content="ok",
+            store=self.store,
+            session_id="test",
+            scratch_key="forced",
+            force=True,
+        )
+        self.assertTrue(result.success)
+
+    def test_evaluate_dry_run(self):
+        """Evaluate returns gate details without promoting."""
+        eval_result = evaluate_for_promotion(
+            content="The fleet uses kimi-k2.5 as the primary model for all agent operations",
+            store=self.store,
+            room="fleet",
+        )
+        self.assertTrue(eval_result["eligible"])
+        self.assertTrue(all(p for p, _ in eval_result["gates"].values()))
+
+    def test_batch_promotion(self):
+        """Batch promotion processes all notes."""
+        notes = {
+            "infra": {"value": "Hermes VPS runs Ubuntu 22.04 with 2 vCPUs and 4GB RAM", "written_at": time.strftime("%Y-%m-%d %H:%M:%S")},
+            "short": {"value": "no", "written_at": time.strftime("%Y-%m-%d %H:%M:%S")},
+            "model": {"value": "The primary local model is gemma4:latest running on Ollama", "written_at": time.strftime("%Y-%m-%d %H:%M:%S")},
+        }
+        results = promote_session_batch(self.store, "batch-session", notes, room="config")
+        promoted = [r for r in results if r.success]
+        rejected = [r for r in results if not r.success]
+        self.assertEqual(len(promoted), 2)
+        self.assertEqual(len(rejected), 1)
+
+    def test_promotion_logged(self):
+        """Successful promotions appear in the audit log."""
+        promote(
+            content="Forge is hosted at forge.alexanderwhitestone.com running Gitea",
+            store=self.store,
+            session_id="log-test",
+            scratch_key="forge",
+            room="infrastructure",
+        )
+        log = self.store.recent_promotions()
+        self.assertTrue(len(log) > 0)
+        self.assertEqual(log[0]["session_id"], "log-test")
+        self.assertEqual(log[0]["scratch_key"], "forge")
+
+
+if __name__ == "__main__":
+    unittest.main()

hermes-sovereign/mempalace/tests/test_wakeup.py

@@ -0,0 +1,100 @@
+"""Tests for wakeup.py.
+
+Refs: Epic #367, Sub-issue #372
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import sys
+import time
+from pathlib import Path
+from unittest.mock import patch, MagicMock
+
+import pytest
+
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", ".."))
+
+from mempalace.wakeup import (
+    palace_wakeup,
+    fleet_status_summary,
+    _load_identity,
+    _palace_context,
+)
+
+
+class TestLoadIdentity:
+    def test_loads_identity(self, tmp_path):
+        f = tmp_path / "identity.txt"
+        f.write_text("I am Timmy. A sovereign AI.")
+        with patch("mempalace.wakeup.IDENTITY_PATH", f):
+            result = _load_identity()
+            assert "Timmy" in result
+
+    def test_missing_identity(self, tmp_path):
+        f = tmp_path / "nope.txt"
+        with patch("mempalace.wakeup.IDENTITY_PATH", f):
+            assert _load_identity() == ""
+
+
+class TestFleetStatus:
+    def test_reads_fleet_json(self, tmp_path):
+        f = tmp_path / "fleet_status.json"
+        f.write_text(json.dumps({
+            "Groq": {"state": "active", "last_seen": "2026-04-07"},
+            "Ezra": {"state": "idle", "last_seen": "2026-04-06"},
+        }))
+        with patch("mempalace.wakeup.FLEET_STATUS_PATH", f):
+            result = fleet_status_summary()
+            assert "Fleet Status" in result
+            assert "Groq" in result
+            assert "active" in result
+
+    def test_missing_fleet_file(self, tmp_path):
+        f = tmp_path / "nope.json"
+        with patch("mempalace.wakeup.FLEET_STATUS_PATH", f):
+            assert fleet_status_summary() == ""
+
+    def test_invalid_json(self, tmp_path):
+        f = tmp_path / "bad.json"
+        f.write_text("not json")
+        with patch("mempalace.wakeup.FLEET_STATUS_PATH", f):
+            assert fleet_status_summary() == ""
+
+
+class TestPalaceWakeup:
+    def test_generates_context_with_identity(self, tmp_path):
+        identity = tmp_path / "identity.txt"
+        identity.write_text("I am Timmy.")
+        cache = tmp_path / "cache.txt"
+        with patch("mempalace.wakeup.IDENTITY_PATH", identity), \
+             patch("mempalace.wakeup.WAKEUP_CACHE_PATH", cache), \
+             patch("mempalace.wakeup._palace_context", return_value=""), \
+             patch("mempalace.wakeup.fleet_status_summary", return_value=""):
+            result = palace_wakeup(force=True)
+            assert "Identity" in result
+            assert "Timmy" in result
+            assert "Session" in result
+
+    def test_uses_cache_when_fresh(self, tmp_path):
+        cache = tmp_path / "cache.txt"
+        cache.write_text("cached wake-up content")
+        # Touch the file so it's fresh
+        with patch("mempalace.wakeup.WAKEUP_CACHE_PATH", cache), \
+             patch("mempalace.wakeup.WAKEUP_CACHE_TTL", 9999):
+            result = palace_wakeup(force=False)
+            assert result == "cached wake-up content"
+
+    def test_force_bypasses_cache(self, tmp_path):
+        cache = tmp_path / "cache.txt"
+        cache.write_text("stale content")
+        identity = tmp_path / "identity.txt"
+        identity.write_text("I am Timmy.")
+        with patch("mempalace.wakeup.WAKEUP_CACHE_PATH", cache), \
+             patch("mempalace.wakeup.IDENTITY_PATH", identity), \
+             patch("mempalace.wakeup._palace_context", return_value=""), \
+             patch("mempalace.wakeup.fleet_status_summary", return_value=""):
+            result = palace_wakeup(force=True)
+            assert "Identity" in result
+            assert "stale content" not in result

hermes-sovereign/mempalace/wakeup.py

@@ -0,0 +1,161 @@
+"""Wake-up Protocol — session start context injection.
+
+Generates 300-900 tokens of context when a new Hermes session starts.
+Loads identity, recent palace context, and fleet status.
+
+Refs: Epic #367, Sub-issue #372
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import subprocess
+import time
+from pathlib import Path
+
+# ---------------------------------------------------------------------------
+# Constants
+# ---------------------------------------------------------------------------
+
+IDENTITY_PATH = Path.home() / ".mempalace" / "identity.txt"
+MEMPALACE_BIN = "/Library/Frameworks/Python.framework/Versions/3.12/bin/mempalace"
+FLEET_STATUS_PATH = Path.home() / ".hermes" / "fleet_status.json"
+WAKEUP_CACHE_PATH = Path.home() / ".hermes" / "last_wakeup.txt"
+WAKEUP_CACHE_TTL = 300  # 5 minutes — don't regenerate if recent
+
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+def _load_identity() -> str:
+    """Read the agent identity file."""
+    try:
+        if IDENTITY_PATH.exists():
+            text = IDENTITY_PATH.read_text(encoding="utf-8").strip()
+            # Cap at ~150 tokens for wake-up brevity
+            words = text.split()
+            if len(words) > 150:
+                text = " ".join(words[:150]) + "..."
+            return text
+    except (OSError, PermissionError):
+        pass
+    return ""
+
+
+def _palace_context() -> str:
+    """Run mempalace wake-up command for recent context. Degrades gracefully."""
+    try:
+        bin_path = MEMPALACE_BIN if os.path.exists(MEMPALACE_BIN) else "mempalace"
+        result = subprocess.run(
+            [bin_path, "wake-up"],
+            capture_output=True,
+            text=True,
+            timeout=10,
+        )
+        if result.returncode == 0 and result.stdout.strip():
+            return result.stdout.strip()
+    except (FileNotFoundError, subprocess.TimeoutExpired, OSError):
+        # ONNX issues (#373) or CLI not available — degrade gracefully
+        pass
+    return ""
+
+
+def fleet_status_summary() -> str:
+    """Read cached fleet status for lightweight session context."""
+    try:
+        if FLEET_STATUS_PATH.exists():
+            data = json.loads(FLEET_STATUS_PATH.read_text(encoding="utf-8"))
+            lines = ["## Fleet Status"]
+
+            if isinstance(data, dict):
+                for agent, status in data.items():
+                    if isinstance(status, dict):
+                        state = status.get("state", "unknown")
+                        last_seen = status.get("last_seen", "?")
+                        lines.append(f"  {agent}: {state} (last: {last_seen})")
+                    else:
+                        lines.append(f"  {agent}: {status}")
+
+            if len(lines) > 1:
+                return "\n".join(lines)
+    except (OSError, json.JSONDecodeError):
+        pass
+    return ""
+
+
+def _check_cache() -> str:
+    """Return cached wake-up if fresh enough."""
+    try:
+        if WAKEUP_CACHE_PATH.exists():
+            age = time.time() - WAKEUP_CACHE_PATH.stat().st_mtime
+            if age < WAKEUP_CACHE_TTL:
+                return WAKEUP_CACHE_PATH.read_text(encoding="utf-8").strip()
+    except OSError:
+        pass
+    return ""
+
+
+def _write_cache(content: str) -> None:
+    """Cache the wake-up content."""
+    try:
+        WAKEUP_CACHE_PATH.parent.mkdir(parents=True, exist_ok=True)
+        WAKEUP_CACHE_PATH.write_text(content, encoding="utf-8")
+    except OSError:
+        pass
+
+
+# ---------------------------------------------------------------------------
+# Main entry point
+# ---------------------------------------------------------------------------
+
+def palace_wakeup(force: bool = False) -> str:
+    """Generate wake-up context for a new session. ~300-900 tokens.
+
+    Args:
+        force: If True, bypass the 5-minute cache and regenerate.
+
+    Returns:
+        Formatted context string suitable for prepending to the system prompt.
+    """
+    # Check cache first (avoids redundant work on rapid session restarts)
+    if not force:
+        cached = _check_cache()
+        if cached:
+            return cached
+
+    parts = []
+
+    # L0: Identity
+    identity = _load_identity()
+    if identity:
+        parts.append(f"## Identity\n{identity}")
+
+    # L1: Recent palace context
+    palace = _palace_context()
+    if palace:
+        parts.append(palace)
+
+    # Fleet status (lightweight)
+    fleet = fleet_status_summary()
+    if fleet:
+        parts.append(fleet)
+
+    # Timestamp
+    parts.append(f"## Session\nWake-up generated: {time.strftime('%Y-%m-%d %H:%M:%S')}")
+
+    content = "\n\n".join(parts)
+
+    # Cache for TTL
+    _write_cache(content)
+
+    return content
+
+
+# ---------------------------------------------------------------------------
+# CLI entry point for testing
+# ---------------------------------------------------------------------------
+
+if __name__ == "__main__":
+    print(palace_wakeup(force=True))

hermes-sovereign/orchestrator/orchestrate.sh

@@ -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" "$@"

hermes-sovereign/orchestrator/orchestrator.py

@@ -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()

playbooks/fleet-guardrails.yaml

@@ -0,0 +1,166 @@
+# fleet-guardrails.yaml
+# =====================
+# Enforceable behaviour boundaries for every agent in the Timmy fleet.
+# Consumed by task_gate.py (pre/post checks) and the orchestrator's
+# dispatch loop. Every rule here is testable — no aspirational prose.
+#
+# Ref: SOUL.md "grounding before generation", Five Wisdoms #345
+
+name: fleet-guardrails
+version: "1.0.0"
+description: >
+  Behaviour constraints that apply to ALL agents regardless of role.
+  These are the non-negotiable rules that task_gate.py enforces
+  before an agent may pick up work and after it claims completion.
+
+# ─── UNIVERSAL CONSTRAINTS ───────────────────────────────────────
+
+constraints:
+
+  # 1. Lane discipline — agents must stay in their lane
+  lane_enforcement:
+    enabled: true
+    source: playbooks/agent-lanes.json
+    on_violation: block_and_notify
+    description: >
+      An agent may only pick up issues tagged for its lane.
+      Cross-lane work requires explicit Timmy approval via
+      issue comment containing 'LANE_OVERRIDE: <agent>'.
+
+  # 2. Branch hygiene — no orphan branches
+  branch_hygiene:
+    enabled: true
+    max_branches_per_agent: 3
+    stale_branch_days: 7
+    naming_pattern: "{agent}/{issue_number}-{slug}"
+    on_violation: warn_then_block
+    description: >
+      Agents must follow branch naming conventions and clean up
+      after merge. No agent may have more than 3 active branches.
+
+  # 3. Issue ownership — no silent takeovers
+  issue_ownership:
+    enabled: true
+    require_assignment_before_work: true
+    max_concurrent_issues: 2
+    on_violation: block_and_notify
+    description: >
+      An agent must be assigned to an issue before creating a
+      branch or PR. No agent may work on more than 2 issues
+      simultaneously to prevent context-switching waste.
+
+  # 4. PR quality — minimum bar before review
+  pr_quality:
+    enabled: true
+    require_linked_issue: true
+    require_passing_ci: true
+    max_files_changed: 30
+    max_diff_lines: 2000
+    require_description: true
+    min_description_length: 50
+    on_violation: block_merge
+    description: >
+      Every PR must link an issue, pass CI, have a meaningful
+      description, and stay within scope. Giant PRs get rejected.
+
+  # 5. Grounding before generation — SOUL.md compliance
+  grounding:
+    enabled: true
+    require_issue_read_before_branch: true
+    require_existing_code_review: true
+    require_soul_md_check: true
+    soul_md_path: SOUL.md
+    on_violation: block_and_notify
+    description: >
+      Before writing any code, the agent must demonstrate it has
+      read the issue, reviewed relevant existing code, and checked
+      SOUL.md for applicable doctrine. No speculative generation.
+
+  # 6. Completion integrity — no phantom completions
+  completion_checks:
+    enabled: true
+    require_test_evidence: true
+    require_ci_green: true
+    require_diff_matches_issue: true
+    require_no_unrelated_changes: true
+    on_violation: revert_and_notify
+    description: >
+      Post-task gate verifies the work actually addresses the
+      issue. Agents cannot close issues without evidence.
+      Unrelated changes in a PR trigger automatic rejection.
+
+  # 7. Communication discipline — no noise
+  communication:
+    enabled: true
+    max_comments_per_issue: 10
+    require_structured_updates: true
+    update_format: "status | what_changed | what_blocked | next_step"
+    prohibit_empty_updates: true
+    on_violation: warn
+    description: >
+      Issue comments must be structured and substantive.
+      Status-only comments without content are rejected.
+      Agents should update, not narrate.
+
+  # 8. Resource awareness — no runaway costs
+  resource_limits:
+    enabled: true
+    max_api_calls_per_task: 100
+    max_llm_tokens_per_task: 500000
+    max_task_duration_minutes: 60
+    on_violation: kill_and_notify
+    description: >
+      Hard limits on compute per task. If an agent hits these
+      limits, the task is killed and flagged for human review.
+      Prevents infinite loops and runaway API spending.
+
+# ─── ESCALATION POLICY ───────────────────────────────────────────
+
+escalation:
+  channels:
+    - gitea_issue_comment
+    - discord_webhook
+  severity_levels:
+    warn:
+      action: post_comment
+      notify: agent_only
+    block:
+      action: prevent_action
+      notify: agent_and_orchestrator
+    block_and_notify:
+      action: prevent_action
+      notify: agent_orchestrator_and_timmy
+    kill_and_notify:
+      action: terminate_task
+      notify: all_including_alexander
+    revert_and_notify:
+      action: revert_changes
+      notify: agent_orchestrator_and_timmy
+
+# ─── AUDIT TRAIL ─────────────────────────────────────────────────
+
+audit:
+  enabled: true
+  log_path: logs/guardrail-violations.jsonl
+  retention_days: 90
+  fields:
+    - timestamp
+    - agent
+    - constraint
+    - violation_type
+    - issue_number
+    - action_taken
+    - resolution
+
+# ─── OVERRIDES ───────────────────────────────────────────────────
+
+overrides:
+  # Only Timmy or Alexander can override guardrails
+  authorized_overriders:
+    - Timmy
+    - Alexander
+  override_mechanism: >
+    Post a comment on the issue with the format:
+    GUARDRAIL_OVERRIDE: <constraint_name> REASON: <explanation>
+  override_expiry_hours: 24
+  require_post_override_review: true