refactor(desktop): tighten right-rail tab close API

Promote closeRightRailTab/closeActiveRightRailTab as the single
public entry point. Drops the activeTabRef + handleCloseDocument
indirection in ChatPreviewRail, the unused $rightRailHasContent
atom, and the legacy dismissFilePreviewTarget alias. -70 LOC.

.dockerignore

@@ -9,6 +9,12 @@ node_modules
 .venv
 **/.venv
 
+# Built artifacts that are regenerated inside the image.  Excluded so local
+# rebuilds on the developer's machine don't invalidate the npm-install layer
+# that now depends on the full ui-tui/packages/hermes-ink/ tree being present.
+ui-tui/dist/
+ui-tui/packages/hermes-ink/dist/
+
 # CI/CD
 .github
 
@@ -19,3 +25,7 @@ node_modules
 
 # Runtime data (bind-mounted at /opt/data; must not leak into build context)
 data/
+
+# Compose/profile runtime state (bind-mounted; avoid ownership/secret issues)
+hermes-config/
+runtime/

.env.example

@@ -384,9 +384,9 @@ IMAGE_TOOLS_DEBUG=false
 # Default STT provider is "local" (faster-whisper) — runs on your machine, no API key needed.
 # Install with: pip install faster-whisper
 # Model downloads automatically on first use (~150 MB for "base").
-# To use cloud providers instead, set GROQ_API_KEY or VOICE_TOOLS_OPENAI_KEY above.
-# Provider priority: local > groq > openai
-# Configure in config.yaml: stt.provider: local | groq | openai
+# To use cloud providers instead, set GROQ_API_KEY, VOICE_TOOLS_OPENAI_KEY, or ELEVENLABS_API_KEY above.
+# Provider priority: local > groq > openai > mistral > xai > elevenlabs
+# Configure in config.yaml: stt.provider: local | groq | openai | mistral | xai | elevenlabs
 
 # =============================================================================
 # STT ADVANCED OVERRIDES (optional)
@@ -394,10 +394,12 @@ IMAGE_TOOLS_DEBUG=false
 # Override default STT models per provider (normally set via stt.model in config.yaml)
 # STT_GROQ_MODEL=whisper-large-v3-turbo
 # STT_OPENAI_MODEL=whisper-1
+# STT_ELEVENLABS_MODEL=scribe_v2
 
 # Override STT provider endpoints (for proxies or self-hosted instances)
 # GROQ_BASE_URL=https://api.groq.com/openai/v1
 # STT_OPENAI_BASE_URL=https://api.openai.com/v1
+# ELEVENLABS_STT_BASE_URL=https://api.elevenlabs.io/v1
 
 # =============================================================================
 # MICROSOFT TEAMS INTEGRATION

.github/dependabot.yml

@@ -0,0 +1,44 @@
+# Dependabot configuration for hermes-agent.
+#
+# Deliberately scoped to github-actions only.
+#
+# We do NOT enable Dependabot for pip / npm / any source-dependency ecosystem
+# because we pin source dependencies exactly (uv.lock, package-lock.json) as
+# part of our supply-chain posture. Automatic version-bump PRs against those
+# pins would undermine the strategy — pins are moved deliberately, after
+# review, not on a schedule.
+#
+# github-actions is the exception: action pins (we use full commit SHAs per
+# supply-chain policy) must be updated when upstream actions publish
+# patches — usually themselves security fixes. Dependabot opens a PR with
+# the new SHA and release notes; we review and merge like any other PR.
+#
+# Security-update PRs for source dependencies (opened ONLY when a CVE is
+# published affecting a currently-pinned version) are enabled separately
+# via the repo's Dependabot security updates setting
+# (Settings → Code security → Dependabot → Dependabot security updates).
+# Those are CVE-only, not schedule-driven, and do not conflict with our
+# pinning strategy — they fire when a pinned version becomes known-bad,
+# which is exactly when we want to move the pin.
+
+version: 2
+updates:
+  - package-ecosystem: "github-actions"
+    directory: "/"
+    schedule:
+      interval: "weekly"
+      day: "monday"
+    open-pull-requests-limit: 5
+    labels:
+      - "dependencies"
+      - "github-actions"
+    commit-message:
+      prefix: "chore(actions)"
+      include: "scope"
+    groups:
+      # Batch routine action bumps into one PR per week to reduce noise.
+      # Security updates still open individually and bypass grouping.
+      actions-minor-patch:
+        update-types:
+          - "minor"
+          - "patch"

.github/workflows/deploy-site.yml

@@ -76,6 +76,16 @@ jobs:
         run: |
           mkdir -p _site/docs
           cp -r website/build/* _site/docs/
+          # llms.txt / llms-full.txt are also published at the site root
+          # (https://hermes-agent.nousresearch.com/llms.txt) because some
+          # agents and IDE plugins probe the classic root-level path rather
+          # than /docs/llms.txt. Same file, two URLs, one source of truth.
+          if [ -f website/build/llms.txt ]; then
+            cp website/build/llms.txt _site/llms.txt
+          fi
+          if [ -f website/build/llms-full.txt ]; then
+            cp website/build/llms-full.txt _site/llms-full.txt
+          fi
 
       - name: Upload artifact
         uses: actions/upload-pages-artifact@56afc609e74202658d3ffba0e8f6dda462b719fa  # v3

.github/workflows/desktop-release.yml

@@ -0,0 +1,341 @@
+name: Desktop Release
+
+on:
+  push:
+    branches: [main]
+  release:
+    types: [published]
+  workflow_dispatch:
+    inputs:
+      channel:
+        description: Release channel to build
+        required: true
+        default: nightly
+        type: choice
+        options:
+          - nightly
+          - stable
+      release_tag:
+        description: "Required when channel=stable (example: v2026.5.5)"
+        required: false
+        type: string
+
+permissions:
+  contents: write
+
+concurrency:
+  group: desktop-release-${{ github.ref }}
+  cancel-in-progress: false
+
+jobs:
+  prepare:
+    if: github.repository == 'NousResearch/hermes-agent'
+    runs-on: ubuntu-latest
+    outputs:
+      channel: ${{ steps.meta.outputs.channel }}
+      release_name: ${{ steps.meta.outputs.release_name }}
+      release_tag: ${{ steps.meta.outputs.release_tag }}
+      version: ${{ steps.meta.outputs.version }}
+      is_stable: ${{ steps.meta.outputs.is_stable }}
+    steps:
+      - id: meta
+        env:
+          EVENT_NAME: ${{ github.event_name }}
+          INPUT_CHANNEL: ${{ github.event.inputs.channel }}
+          INPUT_RELEASE_TAG: ${{ github.event.inputs.release_tag }}
+          RELEASE_TAG_FROM_EVENT: ${{ github.event.release.tag_name }}
+          GITHUB_SHA: ${{ github.sha }}
+        run: |
+          set -euo pipefail
+
+          channel="nightly"
+          release_tag="desktop-nightly"
+          is_stable="false"
+
+          if [[ "$EVENT_NAME" == "release" ]]; then
+            channel="stable"
+            release_tag="$RELEASE_TAG_FROM_EVENT"
+            is_stable="true"
+          elif [[ "$EVENT_NAME" == "workflow_dispatch" && "$INPUT_CHANNEL" == "stable" ]]; then
+            channel="stable"
+            release_tag="$INPUT_RELEASE_TAG"
+            is_stable="true"
+          fi
+
+          if [[ "$channel" == "stable" ]]; then
+            if [[ -z "$release_tag" ]]; then
+              echo "Stable desktop releases require a release tag." >&2
+              exit 1
+            fi
+
+            version="${release_tag#v}"
+            release_name="Hermes Desktop ${release_tag}"
+          else
+            stamp="$(date -u +%Y%m%d)"
+            short_sha="${GITHUB_SHA::7}"
+            version="0.0.0-nightly.${stamp}.${short_sha}"
+            release_name="Hermes Desktop Nightly ${stamp}-${short_sha}"
+          fi
+
+          {
+            echo "channel=$channel"
+            echo "release_name=$release_name"
+            echo "release_tag=$release_tag"
+            echo "version=$version"
+            echo "is_stable=$is_stable"
+          } >> "$GITHUB_OUTPUT"
+
+  build:
+    if: github.repository == 'NousResearch/hermes-agent'
+    needs: prepare
+    strategy:
+      fail-fast: false
+      matrix:
+        include:
+          - platform: mac
+            runner: macos-latest
+            build_args: --mac dmg zip
+          - platform: win
+            runner: windows-latest
+            build_args: --win nsis msi
+    runs-on: ${{ matrix.runner }}
+    env:
+      DESKTOP_CHANNEL: ${{ needs.prepare.outputs.channel }}
+      DESKTOP_VERSION: ${{ needs.prepare.outputs.version }}
+      MAC_CSC_LINK: ${{ secrets.CSC_LINK }}
+      MAC_CSC_KEY_PASSWORD: ${{ secrets.CSC_KEY_PASSWORD }}
+      APPLE_API_KEY: ${{ secrets.APPLE_API_KEY }}
+      APPLE_API_KEY_ID: ${{ secrets.APPLE_API_KEY_ID }}
+      APPLE_API_ISSUER: ${{ secrets.APPLE_API_ISSUER }}
+      WIN_CSC_LINK: ${{ secrets.WIN_CSC_LINK }}
+      WIN_CSC_KEY_PASSWORD: ${{ secrets.WIN_CSC_KEY_PASSWORD }}
+    steps:
+      - uses: actions/checkout@34e114876b0b11c390a56381ad16ebd13914f8d5  # v4
+
+      - uses: actions/setup-node@49933ea5288caeca8642d1e84afbd3f7d6820020  # v4
+        with:
+          node-version: 20
+          cache: npm
+          cache-dependency-path: package-lock.json
+
+      - uses: actions/setup-python@a26af69be951a213d495a4c3e4e4022e16d87065  # v5
+        with:
+          python-version: "3.11"
+
+      - name: Enforce signing gates for stable releases
+        if: needs.prepare.outputs.is_stable == 'true'
+        shell: bash
+        run: |
+          set -euo pipefail
+          missing=()
+
+          if [[ "${{ matrix.platform }}" == "mac" ]]; then
+            [[ -z "${MAC_CSC_LINK:-}" ]] && missing+=("CSC_LINK")
+            [[ -z "${MAC_CSC_KEY_PASSWORD:-}" ]] && missing+=("CSC_KEY_PASSWORD")
+            [[ -z "${APPLE_API_KEY:-}" ]] && missing+=("APPLE_API_KEY")
+            [[ -z "${APPLE_API_KEY_ID:-}" ]] && missing+=("APPLE_API_KEY_ID")
+            [[ -z "${APPLE_API_ISSUER:-}" ]] && missing+=("APPLE_API_ISSUER")
+          else
+            [[ -z "${WIN_CSC_LINK:-}" ]] && missing+=("WIN_CSC_LINK")
+            [[ -z "${WIN_CSC_KEY_PASSWORD:-}" ]] && missing+=("WIN_CSC_KEY_PASSWORD")
+          fi
+
+          if (( ${#missing[@]} > 0 )); then
+            echo "::error::Stable desktop release missing required secrets: ${missing[*]}"
+            exit 1
+          fi
+
+      - name: Install workspace dependencies
+        run: npm ci
+
+      - name: Build bundled TUI payload
+        run: npm --prefix ui-tui run build
+
+      - name: Build desktop renderer
+        run: npm --prefix apps/desktop run build
+
+      - name: Stage Hermes payload
+        run: npm --prefix apps/desktop run stage:hermes
+
+      - name: Map macOS signing credentials
+        if: matrix.platform == 'mac'
+        shell: bash
+        run: |
+          set -euo pipefail
+          has_link=0
+          has_pass=0
+          [[ -n "${MAC_CSC_LINK:-}" ]] && has_link=1
+          [[ -n "${MAC_CSC_KEY_PASSWORD:-}" ]] && has_pass=1
+
+          if [[ $has_link -eq 1 && $has_pass -eq 1 ]]; then
+            echo "CSC_LINK=${MAC_CSC_LINK}" >> "$GITHUB_ENV"
+            echo "CSC_KEY_PASSWORD=${MAC_CSC_KEY_PASSWORD}" >> "$GITHUB_ENV"
+          elif [[ $has_link -eq 1 || $has_pass -eq 1 ]]; then
+            echo "::error::macOS signing secrets are partially configured. Set both CSC_LINK and CSC_KEY_PASSWORD."
+            exit 1
+          fi
+
+      - name: Map Windows signing credentials
+        if: matrix.platform == 'win'
+        shell: bash
+        run: |
+          set -euo pipefail
+          has_link=0
+          has_pass=0
+          [[ -n "${WIN_CSC_LINK:-}" ]] && has_link=1
+          [[ -n "${WIN_CSC_KEY_PASSWORD:-}" ]] && has_pass=1
+
+          if [[ $has_link -eq 1 && $has_pass -eq 1 ]]; then
+            echo "CSC_LINK=${WIN_CSC_LINK}" >> "$GITHUB_ENV"
+            echo "CSC_KEY_PASSWORD=${WIN_CSC_KEY_PASSWORD}" >> "$GITHUB_ENV"
+            echo "CSC_FOR_PULL_REQUEST=true" >> "$GITHUB_ENV"
+          elif [[ $has_link -eq 1 || $has_pass -eq 1 ]]; then
+            echo "::error::Windows signing secrets are partially configured. Set both WIN_CSC_LINK and WIN_CSC_KEY_PASSWORD."
+            exit 1
+          fi
+
+      - name: Build desktop installers
+        shell: bash
+        run: |
+          set -euo pipefail
+          npm --prefix apps/desktop exec electron-builder -- \
+            ${{ matrix.build_args }} \
+            --publish never \
+            --config.extraMetadata.version="${DESKTOP_VERSION}" \
+            --config.extraMetadata.desktopChannel="${DESKTOP_CHANNEL}" \
+            '--config.artifactName=Hermes-${version}-${env.DESKTOP_CHANNEL}-${os}-${arch}.${ext}'
+
+      - name: Notarize and staple macOS DMG
+        if: matrix.platform == 'mac' && needs.prepare.outputs.is_stable == 'true'
+        shell: bash
+        run: |
+          set -euo pipefail
+          dmg_path="$(ls apps/desktop/release/*.dmg | head -n 1)"
+          node apps/desktop/scripts/notarize-artifact.cjs "$dmg_path"
+
+      - name: Validate macOS notarization and Gatekeeper trust
+        if: matrix.platform == 'mac' && needs.prepare.outputs.is_stable == 'true'
+        shell: bash
+        run: |
+          set -euo pipefail
+          app_path="$(ls -d apps/desktop/release/mac*/Hermes.app | head -n 1)"
+          dmg_path="$(ls apps/desktop/release/*.dmg | head -n 1)"
+          xcrun stapler validate "$app_path"
+          xcrun stapler validate "$dmg_path"
+          spctl --assess --type execute --verbose=4 "$app_path"
+
+      - name: Generate desktop checksums
+        shell: bash
+        run: |
+          set -euo pipefail
+          node <<'EOF'
+          const crypto = require('node:crypto')
+          const fs = require('node:fs')
+          const path = require('node:path')
+
+          const releaseDir = path.resolve('apps/desktop/release')
+          const platform = process.env.PLATFORM
+          const extensions = platform === 'mac' ? ['.dmg', '.zip'] : ['.exe', '.msi']
+          const files = fs
+            .readdirSync(releaseDir)
+            .filter(name => extensions.some(ext => name.endsWith(ext)))
+            .sort()
+
+          if (!files.length) {
+            throw new Error(`No release artifacts were produced for ${platform}`)
+          }
+
+          const lines = files.map(name => {
+            const full = path.join(releaseDir, name)
+            const hash = crypto.createHash('sha256').update(fs.readFileSync(full)).digest('hex')
+            return `${hash}  ${name}`
+          })
+          fs.writeFileSync(path.join(releaseDir, `SHA256SUMS-${platform}.txt`), `${lines.join('\n')}\n`)
+          EOF
+        env:
+          PLATFORM: ${{ matrix.platform }}
+
+      - name: Upload packaged desktop artifacts
+        uses: actions/upload-artifact@ea165f8d65b6e75b540449e92b4886f43607fa02  # v4
+        with:
+          name: desktop-${{ matrix.platform }}
+          path: |
+            apps/desktop/release/*.dmg
+            apps/desktop/release/*.zip
+            apps/desktop/release/*.exe
+            apps/desktop/release/*.msi
+            apps/desktop/release/SHA256SUMS-${{ matrix.platform }}.txt
+          if-no-files-found: error
+
+  publish:
+    if: github.repository == 'NousResearch/hermes-agent'
+    needs: [prepare, build]
+    runs-on: ubuntu-latest
+    env:
+      GH_TOKEN: ${{ github.token }}
+      CHANNEL: ${{ needs.prepare.outputs.channel }}
+      RELEASE_NAME: ${{ needs.prepare.outputs.release_name }}
+      RELEASE_TAG: ${{ needs.prepare.outputs.release_tag }}
+    steps:
+      - uses: actions/checkout@34e114876b0b11c390a56381ad16ebd13914f8d5  # v4
+        with:
+          fetch-depth: 0
+
+      - uses: actions/download-artifact@d3f86a106a0bac45b974a628896c90dbdf5c8093  # v4
+        with:
+          pattern: desktop-*
+          merge-multiple: true
+          path: dist/desktop
+
+      - name: Publish desktop assets to GitHub release
+        shell: bash
+        run: |
+          set -euo pipefail
+          shopt -s globstar nullglob
+
+          files=(
+            dist/desktop/**/*.dmg
+            dist/desktop/**/*.zip
+            dist/desktop/**/*.exe
+            dist/desktop/**/*.msi
+            dist/desktop/**/SHA256SUMS-*.txt
+          )
+
+          if (( ${#files[@]} == 0 )); then
+            echo "No desktop artifacts were downloaded for publishing." >&2
+            exit 1
+          fi
+
+          if [[ "$CHANNEL" == "nightly" ]]; then
+            git tag -f "$RELEASE_TAG" "$GITHUB_SHA"
+            git push origin "refs/tags/$RELEASE_TAG" --force
+
+            notes="Automated nightly desktop build from main. This prerelease is replaced on each new run."
+
+            if gh release view "$RELEASE_TAG" >/dev/null 2>&1; then
+              while IFS= read -r asset_name; do
+                gh release delete-asset "$RELEASE_TAG" "$asset_name" --yes
+              done < <(gh release view "$RELEASE_TAG" --json assets -q '.assets[].name')
+
+              gh release edit "$RELEASE_TAG" \
+                --title "$RELEASE_NAME" \
+                --prerelease \
+                --notes "$notes"
+            else
+              gh release create "$RELEASE_TAG" \
+                --target "$GITHUB_SHA" \
+                --title "$RELEASE_NAME" \
+                --notes "$notes" \
+                --prerelease
+            fi
+          else
+            if ! gh release view "$RELEASE_TAG" >/dev/null 2>&1; then
+              notes="Automated desktop artifacts attached by desktop-release workflow."
+              gh release create "$RELEASE_TAG" \
+                --target "$GITHUB_SHA" \
+                --title "$RELEASE_NAME" \
+                --notes "$notes"
+            fi
+          fi
+
+          gh release upload "$RELEASE_TAG" "${files[@]}" --clobber

.github/workflows/nix-lockfile-fix.yml

@@ -6,8 +6,8 @@ on:
     paths:
       - 'ui-tui/package-lock.json'
       - 'ui-tui/package.json'
-      - 'web/package-lock.json'
-      - 'web/package.json'
+      - 'apps/dashboard/package-lock.json'
+      - 'apps/dashboard/package.json'
   workflow_dispatch:
     inputs:
       pr_number:
@@ -28,7 +28,7 @@ concurrency:
 jobs:
   # ── Auto-fix on main ───────────────────────────────────────────────
   # Fires when a push to main touches package.json or package-lock.json
-  # in ui-tui/ or web/. Runs fix-lockfiles and pushes the hash
+  # in ui-tui/ or apps/dashboard/. Runs fix-lockfiles and pushes the hash
   # update commit directly to main so Nix builds never stay broken.
   #
   # Safety invariants:
@@ -110,7 +110,7 @@ jobs:
             # run recompute from the correct package-lock state.
             pkg_changed="$(git diff --name-only "$BASE_SHA"..origin/main -- \
               'ui-tui/package-lock.json' 'ui-tui/package.json' \
-              'web/package-lock.json' 'web/package.json' || true)"
+              'apps/dashboard/package-lock.json' 'apps/dashboard/package.json' || true)"
             if [ -n "$pkg_changed" ]; then
               echo "::warning::Package files changed since hash computation — aborting; a fresh run will recompute"
               exit 0

.github/workflows/osv-scanner.yml

@@ -0,0 +1,67 @@
+name: OSV-Scanner
+
+# Scans lockfiles (uv.lock, package-lock.json) against the OSV vulnerability
+# database. Runs on every PR that touches a lockfile and on a weekly schedule
+# against main.
+#
+# This is detection-only — OSV-Scanner does NOT open PRs or modify pins.
+# It reports known CVEs in currently-pinned dependency versions so we can
+# decide when and how to patch on our own schedule. Our pinning strategy
+# (full SHA / exact version) is preserved; only the notification signal
+# is added.
+#
+# Complements the existing supply-chain-audit.yml workflow (which scans
+# for malicious code patterns in PR diffs) by covering the orthogonal
+# "currently-pinned dep became known-vulnerable" case.
+#
+# Uses Google's officially-recommended reusable workflow, pinned by SHA.
+# Findings land in the repo's Security tab (Code Scanning > OSV-Scanner).
+# fail-on-vuln is disabled so the job does not block merges on pre-existing
+# vulnerabilities in pinned deps that we may need to patch deliberately.
+
+on:
+  pull_request:
+    branches: [main]
+    paths:
+      - 'uv.lock'
+      - 'pyproject.toml'
+      - 'package.json'
+      - 'package-lock.json'
+      - 'ui-tui/package.json'
+      - 'ui-tui/package-lock.json'
+      - 'website/package.json'
+      - 'website/package-lock.json'
+      - '.github/workflows/osv-scanner.yml'
+  push:
+    branches: [main]
+    paths:
+      - 'uv.lock'
+      - 'pyproject.toml'
+      - 'package.json'
+      - 'package-lock.json'
+      - 'ui-tui/package-lock.json'
+      - 'website/package-lock.json'
+  schedule:
+    # Weekly scan against main — catches CVEs published after merge for
+    # deps that haven't changed since.
+    - cron: '0 9 * * 1'
+  workflow_dispatch:
+
+permissions:
+  # Required by the reusable workflow to upload SARIF to the Security tab.
+  actions: read
+  contents: read
+  security-events: write
+
+jobs:
+  scan:
+    name: Scan lockfiles
+    uses: google/osv-scanner-action/.github/workflows/osv-scanner-reusable.yml@c51854704019a247608d928f370c98740469d4b5  # v2.3.5
+    with:
+      # Scan explicit lockfiles rather than recursing, so we only look at
+      # the three sources of truth and skip vendored / test / worktree dirs.
+      scan-args: |-
+        --lockfile=uv.lock
+        --lockfile=ui-tui/package-lock.json
+        --lockfile=website/package-lock.json
+      fail-on-vuln: false

.gitignore

@@ -54,6 +54,10 @@ environments/benchmarks/evals/
 
 # Web UI build output
 hermes_cli/web_dist/
+apps/desktop/build/
+apps/desktop/dist/
+apps/desktop/release/
+apps/desktop/*.tsbuildinfo
 
 # Web UI assets — synced from @nous-research/ui at build time via
 # `npm run sync-assets` (see web/package.json).
@@ -70,3 +74,12 @@ mini-swe-agent/
 result
 website/static/api/skills-index.json
 models-dev-upstream/
+
+# Local editor / agent tooling (machine-specific; keep in global config, not the repo)
+.codex/
+.cursor/
+.gemini/
+.zed/
+.mcp.json
+opencode.json
+config/mcporter.json

AGENTS.md

@@ -2,6 +2,8 @@
 
 Instructions for AI coding assistants and developers working on the hermes-agent codebase.
 
+**Never give up on the right solution.**
+
 ## Development Environment
 
 ```bash
@@ -37,12 +39,17 @@ hermes-agent/
 │   ├── platforms/        # Adapter per platform (telegram, discord, slack, whatsapp,
 │   │                     #   homeassistant, signal, matrix, mattermost, email, sms,
 │   │                     #   dingtalk, wecom, weixin, feishu, qqbot, bluebubbles,
-│   │                     #   webhook, api_server, ...). See ADDING_A_PLATFORM.md.
+│   │                     #   yuanbao, webhook, api_server, ...). See ADDING_A_PLATFORM.md.
 │   └── builtin_hooks/    # Extension point for always-registered gateway hooks (none shipped)
 ├── plugins/              # Plugin system (see "Plugins" section below)
 │   ├── memory/           # Memory-provider plugins (honcho, mem0, supermemory, ...)
 │   ├── context_engine/   # Context-engine plugins
-│   └── <others>/         # Dashboard, image-gen, disk-cleanup, examples, ...
+│   ├── kanban/           # Multi-agent board dispatcher + worker plugin
+│   ├── hermes-achievements/  # Gamified achievement tracking
+│   ├── observability/    # Metrics / traces / logs plugin
+│   ├── image_gen/        # Image-generation providers
+│   └── <others>/         # disk-cleanup, example-dashboard, google_meet, platforms,
+│                         #   spotify, strike-freedom-cockpit, ...
 ├── optional-skills/      # Heavier/niche skills shipped but NOT active by default
 ├── skills/               # Built-in skills bundled with the repo
 ├── ui-tui/               # Ink (React) terminal UI — `hermes --tui`
@@ -53,7 +60,7 @@ hermes-agent/
 ├── environments/         # RL training environments (Atropos)
 ├── scripts/              # run_tests.sh, release.py, auxiliary scripts
 ├── website/              # Docusaurus docs site
-└── tests/                # Pytest suite (~15k tests across ~700 files as of Apr 2026)
+└── tests/                # Pytest suite (~17k tests across ~900 files as of May 2026)
 ```
 
 **User config:** `~/.hermes/config.yaml` (settings), `~/.hermes/.env` (API keys only).
@@ -61,6 +68,29 @@ hermes-agent/
 `gateway.log` when running the gateway. Profile-aware via `get_hermes_home()`.
 Browse with `hermes logs [--follow] [--level ...] [--session ...]`.
 
+## TypeScript Style
+
+Applies to TypeScript across Hermes: desktop, TUI, website, and future TS packages.
+
+- Prefer small nanostores over component state when state is shared, reused, or read by distant UI.
+- Let each feature own its atoms. Chat state belongs near chat, shell state near shell, shared state in `src/store`.
+- Components that render from an atom should use `useStore`. Non-rendering actions should read with `$atom.get()`.
+- Do not pass state through three components when the leaf can subscribe to the atom.
+- Keep persistence beside the atom that owns it.
+- Keep route roots thin. They compose routes and shell; they should not become controllers.
+- No monolithic hooks. A hook should own one narrow job.
+- Prefer colocated action modules over hidden god hooks.
+- If a callback is pure side effect, use the terse void form:
+  `onState={st => void setGatewayState(st)}`.
+- Async UI handlers should make intent explicit:
+  `onClick={() => void save()}`.
+- Prefer interfaces for public props and shared object shapes. Avoid `type X = { ... }` for object props.
+- Extend React primitives for props: `React.ComponentProps<'button'>`, `React.ComponentProps<typeof Dialog>`, `Omit<...>`, `Pick<...>`.
+- Table-driven beats condition ladders when mapping ids, routes, or views.
+- `src/app` owns routes, pages, and page-specific components.
+- `src/store` owns shared atoms.
+- `src/lib` owns shared pure helpers.
+
 ## File Dependency Chain
 
 ```
@@ -244,7 +274,7 @@ npm test          # vitest
 
 The dashboard embeds the real `hermes --tui` — **not** a rewrite.  See `hermes_cli/pty_bridge.py` + the `@app.websocket("/api/pty")` endpoint in `hermes_cli/web_server.py`.
 
-- Browser loads `web/src/pages/ChatPage.tsx`, which mounts xterm.js's `Terminal` with the WebGL renderer, `@xterm/addon-fit` for container-driven resize, and `@xterm/addon-unicode11` for modern wide-character widths.
+- Browser loads `apps/dashboard/src/pages/ChatPage.tsx`, which mounts xterm.js's `Terminal` with the WebGL renderer, `@xterm/addon-fit` for container-driven resize, and `@xterm/addon-unicode11` for modern wide-character widths.
 - `/api/pty?token=…` upgrades to a WebSocket; auth uses the same ephemeral `_SESSION_TOKEN` as REST, via query param (browsers can't set `Authorization` on WS upgrade).
 - The server spawns whatever `hermes --tui` would spawn, through `ptyprocess` (POSIX PTY — WSL works, native Windows does not).
 - Frames: raw PTY bytes each direction; resize via `\x1b[RESIZE:<cols>;<rows>]` intercepted on the server and applied with `TIOCSWINSZ`.
@@ -257,7 +287,16 @@ The dashboard embeds the real `hermes --tui` — **not** a rewrite.  See `hermes
 
 ## Adding New Tools
 
-Requires changes in **2 files**:
+For most custom or local-only tools, do **not** edit Hermes core. Use the plugin
+route instead: create `~/.hermes/plugins/<name>/plugin.yaml` and
+`~/.hermes/plugins/<name>/__init__.py`, then register tools with
+`ctx.register_tool(...)`. Plugin toolsets are discovered automatically and can be
+enabled or disabled without touching `tools/` or `toolsets.py`.
+
+Use the built-in route below only when the user is explicitly contributing a new
+core Hermes tool that should ship in the base system.
+
+Built-in/core tools require changes in **2 files**:
 
 **1. Create `tools/your_tool.py`:**
 ```python
@@ -280,9 +319,9 @@ registry.register(
 )
 ```
 
-**2. Add to `toolsets.py`** — either `_HERMES_CORE_TOOLS` (all platforms) or a new toolset.
+**2. Add to `toolsets.py`** — either `_HERMES_CORE_TOOLS` (all platforms) or a new toolset. **This step is required:** auto-discovery imports the tool and registers its schema, but the tool is only *exposed to an agent* if its name appears in a toolset. `_HERMES_CORE_TOOLS` is not dead code — it's the default bundle every platform's base toolset inherits from.
 
-Auto-discovery: any `tools/*.py` file with a top-level `registry.register()` call is imported automatically — no manual import list to maintain.
+Auto-discovery: any `tools/*.py` file with a top-level `registry.register()` call is imported automatically — no manual import list to maintain. Wiring into a toolset is still a deliberate, manual step.
 
 The registry handles schema collection, dispatch, availability checking, and error wrapping. All handlers MUST return a JSON string.
 
@@ -304,6 +343,22 @@ The registry handles schema collection, dispatch, availability checking, and err
    section is handled automatically by the deep-merge and does NOT require
    a version bump.
 
+### Top-level `config.yaml` sections (non-exhaustive):
+
+`model`, `agent`, `terminal`, `compression`, `display`, `stt`, `tts`,
+`memory`, `security`, `delegation`, `smart_model_routing`, `checkpoints`,
+`auxiliary`, `curator`, `skills`, `gateway`, `logging`, `cron`, `profiles`,
+`plugins`, `honcho`.
+
+`auxiliary` holds per-task overrides for side-LLM work (curator, vision,
+embedding, title generation, session_search, etc.) — each task can pin
+its own provider/model/base_url/max_tokens/reasoning_effort. See
+`agent/auxiliary_client.py::_resolve_auto` for resolution order.
+
+`curator` holds the background skill-maintenance config —
+`enabled`, `interval_hours`, `min_idle_hours`, `stale_after_days`,
+`archive_after_days`, `backup` (nested).
+
 ### .env variables (SECRETS ONLY — API keys, tokens, passwords):
 1. Add to `OPTIONAL_ENV_VARS` in `hermes_cli/config.py` with metadata:
 ```python
@@ -510,11 +565,176 @@ niche skills belong in `optional-skills/`.
 
 ### SKILL.md frontmatter
 
-Standard fields: `name`, `description`, `version`, `platforms`
-(OS-gating list: `[macos]`, `[linux, macos]`, ...),
+Standard fields: `name`, `description`, `version`, `author`, `license`,
+`platforms` (OS-gating list: `[macos]`, `[linux, macos]`, ...),
 `metadata.hermes.tags`, `metadata.hermes.category`,
-`metadata.hermes.config` (config.yaml settings the skill needs — stored
-under `skills.config.<key>`, prompted during setup, injected at load time).
+`metadata.hermes.related_skills`, `metadata.hermes.config` (config.yaml
+settings the skill needs — stored under `skills.config.<key>`, prompted
+during setup, injected at load time).
+
+Top-level `tags:` and `category:` are also accepted and mirrored from
+`metadata.hermes.*` by the loader.
+
+---
+
+## Toolsets
+
+All toolsets are defined in `toolsets.py` as a single `TOOLSETS` dict.
+Each platform's adapter picks a base toolset (e.g. Telegram uses
+`"messaging"`); `_HERMES_CORE_TOOLS` is the default bundle most
+platforms inherit from.
+
+Current toolset keys: `browser`, `clarify`, `code_execution`, `cronjob`,
+`debugging`, `delegation`, `discord`, `discord_admin`, `feishu_doc`,
+`feishu_drive`, `file`, `homeassistant`, `image_gen`, `kanban`, `memory`,
+`messaging`, `moa`, `rl`, `safe`, `search`, `session_search`, `skills`,
+`spotify`, `terminal`, `todo`, `tts`, `video`, `vision`, `web`, `yuanbao`.
+
+Enable/disable per platform via `hermes tools` (the curses UI) or the
+`tools.<platform>.enabled` / `tools.<platform>.disabled` lists in
+`config.yaml`.
+
+---
+
+## Delegation (`delegate_task`)
+
+`tools/delegate_tool.py` spawns a subagent with an isolated
+context + terminal session. Synchronous: the parent waits for the
+child's summary before continuing its own loop — if the parent is
+interrupted, the child is cancelled.
+
+Two shapes:
+
+- **Single:** pass `goal` (+ optional `context`, `toolsets`).
+- **Batch (parallel):** pass `tasks: [...]` — each gets its own subagent
+  running concurrently. Concurrency is capped by
+  `delegation.max_concurrent_children` (default 3).
+
+Roles:
+
+- `role="leaf"` (default) — focused worker. Cannot call `delegate_task`,
+  `clarify`, `memory`, `send_message`, `execute_code`.
+- `role="orchestrator"` — retains `delegate_task` so it can spawn its
+  own workers. Gated by `delegation.orchestrator_enabled` (default true)
+  and bounded by `delegation.max_spawn_depth` (default 2).
+
+Key config knobs (under `delegation:` in `config.yaml`):
+`max_concurrent_children`, `max_spawn_depth`, `child_timeout_seconds`,
+`orchestrator_enabled`, `subagent_auto_approve`, `inherit_mcp_toolsets`,
+`max_iterations`.
+
+Synchronicity rule: delegate_task is **not** durable. For long-running
+work that must outlive the current turn, use `cronjob` or
+`terminal(background=True, notify_on_complete=True)` instead.
+
+---
+
+## Curator (skill lifecycle)
+
+Background skill-maintenance system that tracks usage on agent-created
+skills and auto-archives stale ones. Users never lose skills; archives
+go to `~/.hermes/skills/.archive/` and are restorable.
+
+- **Core:** `agent/curator.py` (review loop, auto-transitions, LLM review
+  prompt) + `agent/curator_backup.py` (pre-run tar.gz snapshots).
+- **CLI:** `hermes_cli/curator.py` wires `hermes curator <verb>` where
+  verbs are: `status`, `run`, `pause`, `resume`, `pin`, `unpin`,
+  `archive`, `restore`, `prune`, `backup`, `rollback`.
+- **Telemetry:** `tools/skill_usage.py` owns the sidecar
+  `~/.hermes/skills/.usage.json` — per-skill `use_count`, `view_count`,
+  `patch_count`, `last_activity_at`, `state` (active / stale /
+  archived), `pinned`.
+
+Invariants:
+- Curator only touches skills with `created_by: "agent"` provenance —
+  bundled + hub-installed skills are off-limits.
+- Never deletes; max destructive action is archive.
+- Pinned skills are exempt from every auto-transition and from the
+  LLM review pass.
+- `skill_manage(action="delete")` refuses pinned skills; patch/edit/
+  write_file/remove_file go through so the agent can keep improving
+  pinned skills.
+
+Config section (`curator:` in `config.yaml`):
+`enabled`, `interval_hours`, `min_idle_hours`, `stale_after_days`,
+`archive_after_days`, `backup.*`.
+
+Full user-facing docs: `website/docs/user-guide/features/curator.md`.
+
+---
+
+## Cron (scheduled jobs)
+
+`cron/jobs.py` (job store) + `cron/scheduler.py` (tick loop). Agents
+schedule jobs via the `cronjob` tool; users via `hermes cron <verb>`
+(`list`, `add`, `edit`, `pause`, `resume`, `run`, `remove`) or the
+`/cron` slash command.
+
+Supported schedule formats:
+- Duration: `"30m"`, `"2h"`, `"1d"`
+- "every" phrase: `"every 2h"`, `"every monday 9am"`
+- 5-field cron expression: `"0 9 * * *"`
+- ISO timestamp (one-shot): `"2026-06-01T09:00:00Z"`
+
+Per-job fields include `skills` (load specific skills), `model` /
+`provider` overrides, `script` (pre-run data-collection script whose
+stdout is injected into the prompt; `no_agent=True` turns the script
+into the entire job), `context_from` (chain job A's last output into
+job B's prompt), `workdir` (run in a specific directory with its
+`AGENTS.md`/`CLAUDE.md` loaded), and multi-platform delivery.
+
+Hardening invariants:
+- **3-minute hard interrupt** on cron sessions — runaway agent loops
+  cannot monopolize the scheduler.
+- Catchup window: half the job's period, clamped to 120s–2h.
+- Grace window: 120s for one-shot jobs whose fire time was missed.
+- File lock at `~/.hermes/cron/.tick.lock` prevents duplicate ticks
+  across processes.
+- Cron sessions pass `skip_memory=True` by default; memory providers
+  intentionally do not run during cron.
+
+Cron deliveries are **not** mirrored into the target gateway session —
+they land in their own cron session with a header/footer frame so the
+main conversation's message-role alternation stays intact.
+
+---
+
+## Kanban (multi-agent work queue)
+
+Durable SQLite-backed board that lets multiple profiles / workers
+collaborate on shared tasks. Users drive it via `hermes kanban <verb>`;
+workers spawned by the dispatcher drive it via a dedicated `kanban_*`
+toolset so their schema footprint is zero when they're not inside a
+kanban task.
+
+- **CLI:** `hermes_cli/kanban.py` wires `hermes kanban` with verbs
+  `init`, `create`, `list` (alias `ls`), `show`, `assign`, `link`,
+  `unlink`, `comment`, `complete`, `block`, `unblock`, `archive`,
+  `tail`, plus less-commonly-used `watch`, `stats`, `runs`, `log`,
+  `assignees`, `heartbeat`, `notify-*`, `dispatch`, `daemon`, `gc`.
+- **Worker toolset:** `tools/kanban_tools.py` exposes `kanban_show`,
+  `kanban_complete`, `kanban_block`, `kanban_heartbeat`, `kanban_comment`,
+  `kanban_create`, `kanban_link` — gated by `HERMES_KANBAN_TASK` so
+  the schema only appears for processes actually running as a worker.
+- **Dispatcher:** long-lived loop that (default every 60s) reclaims
+  stale claims, promotes ready tasks, atomically claims, and spawns
+  assigned profiles. Runs **inside the gateway** by default via
+  `kanban.dispatch_in_gateway: true`.
+- **Plugin assets:** `plugins/kanban/dashboard/` (web UI) +
+  `plugins/kanban/systemd/` (`hermes-kanban-dispatcher.service` for
+  standalone dispatcher deployment).
+
+Isolation model:
+- **Board** is the hard boundary — workers are spawned with
+  `HERMES_KANBAN_BOARD` pinned in their env so they can't see other
+  boards.
+- **Tenant** is a soft namespace *within* a board — one specialist
+  fleet can serve multiple businesses with workspace-path + memory-key
+  isolation.
+- After ~5 consecutive spawn failures on the same task the dispatcher
+  auto-blocks it to prevent spin loops.
+
+Full user-facing docs: `website/docs/user-guide/features/kanban.md`.
 
 ---
 

Dockerfile

@@ -28,10 +28,26 @@ WORKDIR /opt/hermes
 # ---------- Layer-cached dependency install ----------
 # Copy only package manifests first so npm install + Playwright are cached
 # unless the lockfiles themselves change.
+#
+# ui-tui/packages/hermes-ink/ is copied IN FULL (not just its manifests)
+# because it is referenced as a `file:` workspace dependency from
+# ui-tui/package.json.  Copying the tree up front lets npm resolve the
+# workspace to real content instead of stopping at a bare package.json.
 COPY package.json package-lock.json ./
 COPY web/package.json web/package-lock.json web/
 COPY ui-tui/package.json ui-tui/package-lock.json ui-tui/
-COPY ui-tui/packages/hermes-ink/package.json ui-tui/packages/hermes-ink/package-lock.json ui-tui/packages/hermes-ink/
+COPY ui-tui/packages/hermes-ink/ ui-tui/packages/hermes-ink/
+
+# `npm_config_install_links=false` forces npm to install `file:` deps as
+# symlinks (the npm 10+ default) even on Debian's older bundled npm 9.x,
+# which defaults to `install-links=true` and installs file deps as *copies*.
+# The host-side package-lock.json is generated with a newer npm that uses
+# symlinks, so an install-as-copy produces a hidden node_modules/.package-lock.json
+# that permanently disagrees with the root lock on the @hermes/ink entry.
+# That disagreement trips the TUI launcher's `_tui_need_npm_install()`
+# check on every startup and triggers a runtime `npm install` that then
+# fails with EACCES (node_modules/ is root-owned from build time).
+ENV npm_config_install_links=false
 
 RUN npm install --prefer-offline --no-audit && \
     npx playwright install --with-deps chromium --only-shell && \
@@ -45,13 +61,7 @@ COPY --chown=hermes:hermes . .
 
 # Build browser dashboard and terminal UI assets.
 RUN cd web && npm run build && \
-    cd ../ui-tui && npm run build && \
-    rm -rf node_modules/@hermes/ink && \
-    rm -rf packages/hermes-ink/node_modules && \
-    cp -R packages/hermes-ink node_modules/@hermes/ink && \
-    npm install --omit=dev --prefer-offline --no-audit --prefix node_modules/@hermes/ink && \
-    rm -rf node_modules/@hermes/ink/node_modules/react && \
-    node --input-type=module -e "await import('@hermes/ink')"
+    cd ../ui-tui && npm run build
 
 # ---------- Permissions ----------
 # Make install dir world-readable so any HERMES_UID can read it at runtime.

acp_adapter/server.py

@@ -4,6 +4,7 @@ from __future__ import annotations
 
 import asyncio
 import contextvars
+import json
 import logging
 import os
 from collections import defaultdict, deque
@@ -47,6 +48,7 @@ from acp.schema import (
     TextContentBlock,
     UnstructuredCommandInput,
     Usage,
+    UsageUpdate,
     UserMessageChunk,
 )
 
@@ -65,6 +67,7 @@ from acp_adapter.events import (
 )
 from acp_adapter.permissions import make_approval_callback
 from acp_adapter.session import SessionManager, SessionState, _expand_acp_enabled_toolsets
+from acp_adapter.tools import build_tool_complete, build_tool_start
 
 logger = logging.getLogger(__name__)
 
@@ -164,6 +167,8 @@ class HermesACPAgent(acp.Agent):
         "context": "Show conversation context info",
         "reset": "Clear conversation history",
         "compact": "Compress conversation context",
+        "steer": "Inject guidance into the currently running agent turn",
+        "queue": "Queue a prompt to run after the current turn finishes",
         "version": "Show Hermes version",
     }
 
@@ -193,6 +198,16 @@ class HermesACPAgent(acp.Agent):
             "name": "compact",
             "description": "Compress conversation context",
         },
+        {
+            "name": "steer",
+            "description": "Inject guidance into the currently running agent turn",
+            "input_hint": "guidance for the active turn",
+        },
+        {
+            "name": "queue",
+            "description": "Queue a prompt to run after the current turn finishes",
+            "input_hint": "prompt to run next",
+        },
         {
             "name": "version",
             "description": "Show Hermes version",
@@ -303,6 +318,66 @@ class HermesACPAgent(acp.Agent):
 
         return target_provider, new_model
 
+    @staticmethod
+    def _build_usage_update(state: SessionState) -> UsageUpdate | None:
+        """Build ACP native context-usage data for clients like Zed.
+
+        Zed's circular context indicator is driven by ACP ``usage_update``
+        session updates: ``size`` is the model context window and ``used`` is
+        the current request pressure.  Hermes estimates ``used`` from the same
+        buckets it sends to providers: system prompt, conversation history, and
+        tool schemas.
+        """
+        agent = state.agent
+        compressor = getattr(agent, "context_compressor", None)
+        size = int(getattr(compressor, "context_length", 0) or 0)
+        if size <= 0:
+            return None
+
+        try:
+            from agent.model_metadata import estimate_request_tokens_rough
+
+            used = estimate_request_tokens_rough(
+                state.history,
+                system_prompt=getattr(agent, "_cached_system_prompt", "") or "",
+                tools=getattr(agent, "tools", None) or None,
+            )
+        except Exception:
+            logger.debug("Could not estimate ACP native context usage", exc_info=True)
+            used = int(getattr(compressor, "last_prompt_tokens", 0) or 0)
+
+        return UsageUpdate(
+            session_update="usage_update",
+            size=max(size, 0),
+            used=max(used, 0),
+        )
+
+    async def _send_usage_update(self, state: SessionState) -> None:
+        """Send ACP native context usage to the connected client."""
+        if not self._conn:
+            return
+        update = self._build_usage_update(state)
+        if update is None:
+            return
+        try:
+            await self._conn.session_update(
+                session_id=state.session_id,
+                update=update,
+            )
+        except Exception:
+            logger.warning(
+                "Failed to send ACP usage update for session %s",
+                state.session_id,
+                exc_info=True,
+            )
+
+    def _schedule_usage_update(self, state: SessionState) -> None:
+        """Schedule native context indicator refresh after ACP responses."""
+        if not self._conn:
+            return
+        loop = asyncio.get_running_loop()
+        loop.call_soon(asyncio.create_task, self._send_usage_update(state))
+
     async def _register_session_mcp_servers(
         self,
         state: SessionState,
@@ -473,37 +548,99 @@ class HermesACPAgent(acp.Agent):
             )
         return None
 
+    @staticmethod
+    def _history_tool_call_name_args(tool_call: dict[str, Any]) -> tuple[str, dict[str, Any]]:
+        """Extract function name/arguments from an OpenAI-style tool_call."""
+        function = tool_call.get("function") if isinstance(tool_call.get("function"), dict) else {}
+        name = str(function.get("name") or tool_call.get("name") or "unknown_tool")
+        raw_args = function.get("arguments") or tool_call.get("arguments") or tool_call.get("args") or {}
+        if isinstance(raw_args, str):
+            try:
+                parsed = json.loads(raw_args)
+            except Exception:
+                parsed = {"raw": raw_args}
+            raw_args = parsed
+        if not isinstance(raw_args, dict):
+            raw_args = {}
+        return name, raw_args
+
+    @staticmethod
+    def _history_tool_call_id(tool_call: dict[str, Any]) -> str:
+        """Return the stable provider tool call id for ACP history replay."""
+        return str(
+            tool_call.get("id")
+            or tool_call.get("call_id")
+            or tool_call.get("tool_call_id")
+            or ""
+        ).strip()
+
     async def _replay_session_history(self, state: SessionState) -> None:
         """Send persisted user/assistant history to clients during session/load.
 
         Zed's ACP history UI calls ``session/load`` after the user picks an item
         from the Agents sidebar. The agent must then replay the full conversation
-        as ``user_message_chunk`` / ``agent_message_chunk`` notifications; merely
-        restoring server-side state makes Hermes remember context, but leaves the
-        editor looking like a clean thread.
+        as user/assistant chunks plus reconstructed tool-call start/completion
+        notifications; merely restoring server-side state makes Hermes remember
+        context, but leaves the editor looking like a clean thread.
         """
         if not self._conn or not state.history:
             return
 
-        for message in state.history:
-            role = str(message.get("role") or "")
-            if role not in {"user", "assistant"}:
-                continue
-            text = self._history_message_text(message)
-            if not text:
-                continue
-            update = self._history_message_update(role=role, text=text)
-            if update is None:
-                continue
+        active_tool_calls: dict[str, tuple[str, dict[str, Any]]] = {}
+
+        async def _send(update: Any) -> bool:
             try:
                 await self._conn.session_update(session_id=state.session_id, update=update)
+                return True
             except Exception:
                 logger.warning(
                     "Failed to replay ACP history for session %s",
                     state.session_id,
                     exc_info=True,
                 )
-                return
+                return False
+
+        for message in state.history:
+            role = str(message.get("role") or "")
+
+            if role in {"user", "assistant"}:
+                text = self._history_message_text(message)
+                if text:
+                    update = self._history_message_update(role=role, text=text)
+                    if update is not None and not await _send(update):
+                        return
+
+            if role == "assistant" and isinstance(message.get("tool_calls"), list):
+                for tool_call in message["tool_calls"]:
+                    if not isinstance(tool_call, dict):
+                        continue
+                    tool_call_id = self._history_tool_call_id(tool_call)
+                    if not tool_call_id:
+                        continue
+                    tool_name, args = self._history_tool_call_name_args(tool_call)
+                    active_tool_calls[tool_call_id] = (tool_name, args)
+                    if not await _send(build_tool_start(tool_call_id, tool_name, args)):
+                        return
+                continue
+
+            if role == "tool":
+                tool_call_id = str(message.get("tool_call_id") or "").strip()
+                tool_name = str(message.get("tool_name") or "").strip()
+                function_args: dict[str, Any] | None = None
+                if tool_call_id in active_tool_calls:
+                    tool_name, function_args = active_tool_calls.pop(tool_call_id)
+                if not tool_call_id or not tool_name:
+                    continue
+                result = message.get("content")
+                if not await _send(
+                    build_tool_complete(
+                        tool_call_id,
+                        tool_name,
+                        result=result if isinstance(result, str) else None,
+                        function_args=function_args,
+                    )
+                ):
+                    return
 
     async def new_session(
         self,
@@ -515,11 +652,24 @@ class HermesACPAgent(acp.Agent):
         await self._register_session_mcp_servers(state, mcp_servers)
         logger.info("New session %s (cwd=%s)", state.session_id, cwd)
         self._schedule_available_commands_update(state.session_id)
+        self._schedule_usage_update(state)
         return NewSessionResponse(
             session_id=state.session_id,
             models=self._build_model_state(state),
         )
 
+    def _schedule_history_replay(self, state: SessionState) -> None:
+        """Replay persisted history after session/load or session/resume returns.
+
+        Zed only attaches streamed transcript/tool updates once the load/resume
+        response has completed. Sending replay notifications while the request is
+        still in-flight can make the server look correct in logs while the editor
+        drops or fails to attach the tool-call history.
+        """
+        loop = asyncio.get_running_loop()
+        replay_coro = self._replay_session_history(state)
+        loop.call_soon(asyncio.create_task, replay_coro)
+
     async def load_session(
         self,
         cwd: str,
@@ -533,8 +683,9 @@ class HermesACPAgent(acp.Agent):
             return None
         await self._register_session_mcp_servers(state, mcp_servers)
         logger.info("Loaded session %s", session_id)
-        await self._replay_session_history(state)
+        self._schedule_history_replay(state)
         self._schedule_available_commands_update(session_id)
+        self._schedule_usage_update(state)
         return LoadSessionResponse(models=self._build_model_state(state))
 
     async def resume_session(
@@ -550,13 +701,17 @@ class HermesACPAgent(acp.Agent):
             state = self.session_manager.create_session(cwd=cwd)
         await self._register_session_mcp_servers(state, mcp_servers)
         logger.info("Resumed session %s", state.session_id)
-        await self._replay_session_history(state)
+        self._schedule_history_replay(state)
         self._schedule_available_commands_update(state.session_id)
+        self._schedule_usage_update(state)
         return ResumeSessionResponse(models=self._build_model_state(state))
 
     async def cancel(self, session_id: str, **kwargs: Any) -> None:
         state = self.session_manager.get_session(session_id)
         if state and state.cancel_event:
+            with state.runtime_lock:
+                if state.is_running and state.current_prompt_text:
+                    state.interrupted_prompt_text = state.current_prompt_text
             state.cancel_event.set()
             try:
                 if getattr(state, "agent", None) and hasattr(state.agent, "interrupt"):
@@ -654,6 +809,39 @@ class HermesACPAgent(acp.Agent):
         if not has_content:
             return PromptResponse(stop_reason="end_turn")
 
+        # /steer on an idle session has no in-flight tool call to inject into.
+        # Rewrite it so the payload runs as a normal user prompt, matching the
+        # gateway's behavior (gateway/run.py ~L4898). Two sub-cases:
+        #   1. Zed-interrupt salvage — a prior prompt was cancelled by the
+        #      client right before /steer arrived; replay it with the steer
+        #      text attached as explicit correction/guidance so the user's
+        #      in-flight work isn't lost.
+        #   2. Plain idle — no prior work to salvage; just run the steer
+        #      payload as a regular prompt. Without this, _cmd_steer would
+        #      silently append to state.queued_prompts and respond with
+        #      "No active turn — queued for the next turn", which looks like
+        #      /queue even though the user never typed /queue.
+        if isinstance(user_content, str) and user_text.startswith("/steer"):
+            steer_text = user_text.split(maxsplit=1)[1].strip() if len(user_text.split(maxsplit=1)) > 1 else ""
+            interrupted_prompt = ""
+            rewrite_idle = False
+            with state.runtime_lock:
+                if not state.is_running and steer_text:
+                    if state.interrupted_prompt_text:
+                        interrupted_prompt = state.interrupted_prompt_text
+                        state.interrupted_prompt_text = ""
+                    else:
+                        rewrite_idle = True
+            if interrupted_prompt:
+                user_text = (
+                    f"{interrupted_prompt}\n\n"
+                    f"User correction/guidance after interrupt: {steer_text}"
+                )
+                user_content = user_text
+            elif rewrite_idle:
+                user_text = steer_text
+                user_content = steer_text
+
         # Intercept slash commands — handle locally without calling the LLM.
         # Slash commands are text-only; if the client included images/resources,
         # send the whole multimodal prompt to the agent instead of treating it as
@@ -664,8 +852,27 @@ class HermesACPAgent(acp.Agent):
                 if self._conn:
                     update = acp.update_agent_message_text(response_text)
                     await self._conn.session_update(session_id, update)
+                    await self._send_usage_update(state)
                 return PromptResponse(stop_reason="end_turn")
 
+        # If Zed sends another regular prompt while the same ACP session is
+        # still running, queue it instead of racing two AIAgent loops against
+        # the same state.history. /steer and /queue are handled above and can
+        # land immediately.
+        with state.runtime_lock:
+            if state.is_running:
+                queued_text = user_text or "[Image attachment]"
+                state.queued_prompts.append(queued_text)
+                depth = len(state.queued_prompts)
+                if self._conn:
+                    update = acp.update_agent_message_text(
+                        f"Queued for the next turn. ({depth} queued)"
+                    )
+                    await self._conn.session_update(session_id, update)
+                return PromptResponse(stop_reason="end_turn")
+            state.is_running = True
+            state.current_prompt_text = user_text or "[Image attachment]"
+
         logger.info("Prompt on session %s: %s", session_id, user_text[:100])
 
         conn = self._conn
@@ -678,24 +885,37 @@ class HermesACPAgent(acp.Agent):
         tool_call_meta: dict[str, dict[str, Any]] = {}
         previous_approval_cb = None
 
+        streamed_message = False
+
         if conn:
             tool_progress_cb = make_tool_progress_cb(conn, session_id, loop, tool_call_ids, tool_call_meta)
-            thinking_cb = make_thinking_cb(conn, session_id, loop)
+            reasoning_cb = make_thinking_cb(conn, session_id, loop)
             step_cb = make_step_cb(conn, session_id, loop, tool_call_ids, tool_call_meta)
             message_cb = make_message_cb(conn, session_id, loop)
+
+            def stream_delta_cb(text: str) -> None:
+                nonlocal streamed_message
+                if text:
+                    streamed_message = True
+                message_cb(text)
+
             approval_cb = make_approval_callback(conn.request_permission, loop, session_id)
         else:
             tool_progress_cb = None
-            thinking_cb = None
+            reasoning_cb = None
             step_cb = None
-            message_cb = None
+            stream_delta_cb = None
             approval_cb = None
 
         agent = state.agent
         agent.tool_progress_callback = tool_progress_cb
-        agent.thinking_callback = thinking_cb
+        # ACP thought panes should not receive Hermes' local kawaii waiting/status
+        # updates. Route provider/model reasoning deltas instead; if the provider
+        # emits no reasoning, Zed should not get a fake "thinking" accordion.
+        agent.thinking_callback = None
+        agent.reasoning_callback = reasoning_cb
         agent.step_callback = step_cb
-        agent.message_callback = message_cb
+        agent.stream_delta_callback = stream_delta_cb
 
         # Approval callback is per-thread (thread-local, GHSA-qg5c-hvr5-hjgr).
         # Set it INSIDE _run_agent so the TLS write happens in the executor
@@ -777,6 +997,9 @@ class HermesACPAgent(acp.Agent):
             result = await loop.run_in_executor(_executor, ctx.run, _run_agent)
         except Exception:
             logger.exception("Executor error for session %s", session_id)
+            with state.runtime_lock:
+                state.is_running = False
+                state.current_prompt_text = ""
             return PromptResponse(stop_reason="end_turn")
 
         if result.get("messages"):
@@ -798,10 +1021,32 @@ class HermesACPAgent(acp.Agent):
                 )
             except Exception:
                 logger.debug("Failed to auto-title ACP session %s", session_id, exc_info=True)
-        if final_response and conn:
+        if final_response and conn and not streamed_message:
             update = acp.update_agent_message_text(final_response)
             await conn.session_update(session_id, update)
 
+        # Mark this turn idle before draining queued work so recursive prompt()
+        # calls can acquire the session. Queued turns are intentionally run as
+        # normal follow-up user prompts, preserving role alternation and history.
+        with state.runtime_lock:
+            state.is_running = False
+            state.current_prompt_text = ""
+
+        while True:
+            with state.runtime_lock:
+                if not state.queued_prompts:
+                    break
+                next_prompt = state.queued_prompts.pop(0)
+            if conn:
+                await conn.session_update(
+                    session_id,
+                    acp.update_user_message_text(next_prompt),
+                )
+            await self.prompt(
+                prompt=[TextContentBlock(type="text", text=next_prompt)],
+                session_id=session_id,
+            )
+
         usage = None
         if any(result.get(key) is not None for key in ("prompt_tokens", "completion_tokens", "total_tokens")):
             usage = Usage(
@@ -812,6 +1057,8 @@ class HermesACPAgent(acp.Agent):
                 cached_read_tokens=result.get("cache_read_tokens"),
             )
 
+        await self._send_usage_update(state)
+
         stop_reason = "cancelled" if state.cancel_event and state.cancel_event.is_set() else "end_turn"
         return PromptResponse(stop_reason=stop_reason, usage=usage)
 
@@ -879,6 +1126,8 @@ class HermesACPAgent(acp.Agent):
             "context": self._cmd_context,
             "reset": self._cmd_reset,
             "compact": self._cmd_compact,
+            "steer": self._cmd_steer,
+            "queue": self._cmd_queue,
             "version": self._cmd_version,
         }.get(cmd)
 
@@ -942,22 +1191,84 @@ class HermesACPAgent(acp.Agent):
             return f"Could not list tools: {e}"
 
     def _cmd_context(self, args: str, state: SessionState) -> str:
+        """Show ACP session context pressure and compression guidance."""
         n_messages = len(state.history)
-        if n_messages == 0:
-            return "Conversation is empty (no messages yet)."
-        # Count by role
+
+        # Count by role.
         roles: dict[str, int] = {}
         for msg in state.history:
             role = msg.get("role", "unknown")
             roles[role] = roles.get(role, 0) + 1
+
+        agent = state.agent
+        model = state.model or getattr(agent, "model", "")
+        provider = getattr(agent, "provider", None) or "auto"
+        compressor = getattr(agent, "context_compressor", None)
+        context_length = int(getattr(compressor, "context_length", 0) or 0)
+        threshold_tokens = int(getattr(compressor, "threshold_tokens", 0) or 0)
+
+        try:
+            from agent.model_metadata import estimate_request_tokens_rough
+
+            system_prompt = getattr(agent, "_cached_system_prompt", "") or ""
+            tools = getattr(agent, "tools", None) or None
+            approx_tokens = estimate_request_tokens_rough(
+                state.history,
+                system_prompt=system_prompt,
+                tools=tools,
+            )
+        except Exception:
+            logger.debug("Could not estimate ACP context usage", exc_info=True)
+            approx_tokens = 0
+
+        if threshold_tokens <= 0 and context_length > 0:
+            threshold_tokens = int(context_length * 0.80)
+
         lines = [
-            f"Conversation: {n_messages} messages",
+            f"Conversation: {n_messages} messages"
+            if n_messages
+            else "Conversation is empty (no messages yet).",
             f"  user: {roles.get('user', 0)}, assistant: {roles.get('assistant', 0)}, "
             f"tool: {roles.get('tool', 0)}, system: {roles.get('system', 0)}",
         ]
-        model = state.model or getattr(state.agent, "model", "")
         if model:
             lines.append(f"Model: {model}")
+        lines.append(f"Provider: {provider}")
+
+        if approx_tokens > 0:
+            if context_length > 0:
+                usage_pct = (approx_tokens / context_length) * 100
+                lines.append(
+                    f"Context usage: ~{approx_tokens:,} / {context_length:,} tokens ({usage_pct:.1f}%)"
+                )
+            else:
+                lines.append(f"Context usage: ~{approx_tokens:,} tokens")
+
+        if threshold_tokens > 0:
+            if approx_tokens > 0:
+                threshold_pct = (threshold_tokens / context_length) * 100 if context_length > 0 else 0
+                remaining = max(threshold_tokens - approx_tokens, 0)
+                if approx_tokens >= threshold_tokens:
+                    lines.append(
+                        f"Compression: due now (threshold ~{threshold_tokens:,}"
+                        + (f", {threshold_pct:.0f}%" if threshold_pct else "")
+                        + "). Run /compact."
+                    )
+                else:
+                    lines.append(
+                        f"Compression: ~{remaining:,} tokens until threshold "
+                        f"(~{threshold_tokens:,}"
+                        + (f", {threshold_pct:.0f}%" if threshold_pct else "")
+                        + ")."
+                    )
+            else:
+                lines.append(f"Compression threshold: ~{threshold_tokens:,} tokens")
+
+        if getattr(agent, "compression_enabled", True) is False:
+            lines.append("Compression is disabled for this agent.")
+        else:
+            lines.append("Tip: run /compact to compress manually before the threshold.")
+
         return "\n".join(lines)
 
     def _cmd_reset(self, args: str, state: SessionState) -> str:
@@ -975,10 +1286,16 @@ class HermesACPAgent(acp.Agent):
             if not hasattr(agent, "_compress_context"):
                 return "Context compression not available for this agent."
 
-            from agent.model_metadata import estimate_messages_tokens_rough
+            from agent.model_metadata import estimate_request_tokens_rough
 
             original_count = len(state.history)
-            approx_tokens = estimate_messages_tokens_rough(state.history)
+            # Include system prompt + tool schemas so the figure reflects real
+            # request pressure, not a transcript-only underestimate (#6217).
+            _sys_prompt = getattr(agent, "_cached_system_prompt", "") or ""
+            _tools = getattr(agent, "tools", None) or None
+            approx_tokens = estimate_request_tokens_rough(
+                state.history, system_prompt=_sys_prompt, tools=_tools
+            )
             original_session_db = getattr(agent, "_session_db", None)
 
             try:
@@ -998,7 +1315,13 @@ class HermesACPAgent(acp.Agent):
             self.session_manager.save_session(state.session_id)
 
             new_count = len(state.history)
-            new_tokens = estimate_messages_tokens_rough(state.history)
+            _sys_prompt_after = getattr(agent, "_cached_system_prompt", "") or _sys_prompt
+            _tools_after = getattr(agent, "tools", None) or _tools
+            new_tokens = estimate_request_tokens_rough(
+                state.history,
+                system_prompt=_sys_prompt_after,
+                tools=_tools_after,
+            )
             return (
                 f"Context compressed: {original_count} -> {new_count} messages\n"
                 f"~{approx_tokens:,} -> ~{new_tokens:,} tokens"
@@ -1006,6 +1329,34 @@ class HermesACPAgent(acp.Agent):
         except Exception as e:
             return f"Compression failed: {e}"
 
+    def _cmd_steer(self, args: str, state: SessionState) -> str:
+        steer_text = args.strip()
+        if not steer_text:
+            return "Usage: /steer <guidance>"
+
+        if state.is_running and hasattr(state.agent, "steer"):
+            try:
+                if state.agent.steer(steer_text):
+                    preview = steer_text[:80] + ("..." if len(steer_text) > 80 else "")
+                    return f"⏩ Steer queued for the active turn: {preview}"
+            except Exception as exc:
+                logger.warning("ACP steer failed for session %s: %s", state.session_id, exc)
+                return f"⚠️ Steer failed: {exc}"
+
+        with state.runtime_lock:
+            state.queued_prompts.append(steer_text)
+            depth = len(state.queued_prompts)
+        return f"No active turn — queued for the next turn. ({depth} queued)"
+
+    def _cmd_queue(self, args: str, state: SessionState) -> str:
+        queued_text = args.strip()
+        if not queued_text:
+            return "Usage: /queue <prompt>"
+        with state.runtime_lock:
+            state.queued_prompts.append(queued_text)
+            depth = len(state.queued_prompts)
+        return f"Queued for the next turn. ({depth} queued)"
+
     def _cmd_version(self, args: str, state: SessionState) -> str:
         return f"Hermes Agent v{HERMES_VERSION}"
 

acp_adapter/session.py

@@ -26,6 +26,33 @@ from typing import Any, Dict, List, Optional
 logger = logging.getLogger(__name__)
 
 
+def _win_path_to_wsl(path: str) -> str | None:
+    """Convert a Windows drive path to its WSL /mnt/<drive>/... equivalent."""
+    match = re.match(r"^([A-Za-z]):[\\/](.*)$", path)
+    if not match:
+        return None
+    drive = match.group(1).lower()
+    tail = match.group(2).replace("\\", "/")
+    return f"/mnt/{drive}/{tail}"
+
+
+def _translate_acp_cwd(cwd: str) -> str:
+    """Translate Windows ACP cwd values when Hermes itself is running in WSL.
+
+    Windows ACP clients can launch ``hermes acp`` inside WSL while still sending
+    editor workspaces as Windows drive paths such as ``E:\\Projects``. Store
+    and execute against the WSL mount path so agents, tools, and persisted ACP
+    sessions all agree on the usable workspace. Native Linux/macOS keeps the
+    original cwd unchanged.
+    """
+    from hermes_constants import is_wsl
+
+    if not is_wsl():
+        return cwd
+    translated = _win_path_to_wsl(str(cwd))
+    return translated if translated is not None else cwd
+
+
 def _normalize_cwd_for_compare(cwd: str | None) -> str:
     raw = str(cwd or ".").strip()
     if not raw:
@@ -34,11 +61,9 @@ def _normalize_cwd_for_compare(cwd: str | None) -> str:
 
     # Normalize Windows drive paths into the equivalent WSL mount form so
     # ACP history filters match the same workspace across Windows and WSL.
-    match = re.match(r"^([A-Za-z]):[\\/](.*)$", expanded)
-    if match:
-        drive = match.group(1).lower()
-        tail = match.group(2).replace("\\", "/")
-        expanded = f"/mnt/{drive}/{tail}"
+    translated = _win_path_to_wsl(expanded)
+    if translated is not None:
+        expanded = translated
     elif re.match(r"^/mnt/[A-Za-z]/", expanded):
         expanded = f"/mnt/{expanded[5].lower()}/{expanded[7:]}"
 
@@ -96,12 +121,18 @@ def _acp_stderr_print(*args, **kwargs) -> None:
 
 
 def _register_task_cwd(task_id: str, cwd: str) -> None:
-    """Bind a task/session id to the editor's working directory for tools."""
+    """Bind a task/session id to the editor's working directory for tools.
+
+    Zed can launch Hermes from a Windows workspace while the ACP process runs
+    inside WSL. In that case ACP sends cwd as e.g. ``E:\\Projects\\POTI``;
+    local tools need the WSL mount equivalent or subprocess creation fails
+    before the command can run.
+    """
     if not task_id:
         return
     try:
         from tools.terminal_tool import register_task_env_overrides
-        register_task_env_overrides(task_id, {"cwd": cwd})
+        register_task_env_overrides(task_id, {"cwd": _translate_acp_cwd(cwd)})
     except Exception:
         logger.debug("Failed to register ACP task cwd override", exc_info=True)
 
@@ -145,6 +176,11 @@ class SessionState:
     model: str = ""
     history: List[Dict[str, Any]] = field(default_factory=list)
     cancel_event: Any = None  # threading.Event
+    is_running: bool = False
+    queued_prompts: List[str] = field(default_factory=list)
+    runtime_lock: Any = field(default_factory=Lock)
+    current_prompt_text: str = ""
+    interrupted_prompt_text: str = ""
 
 
 class SessionManager:
@@ -175,6 +211,7 @@ class SessionManager:
         """Create a new session with a unique ID and a fresh AIAgent."""
         import threading
 
+        cwd = _translate_acp_cwd(cwd)
         session_id = str(uuid.uuid4())
         agent = self._make_agent(session_id=session_id, cwd=cwd)
         state = SessionState(
@@ -217,6 +254,7 @@ class SessionManager:
         """Deep-copy a session's history into a new session."""
         import threading
 
+        cwd = _translate_acp_cwd(cwd)
         original = self.get_session(session_id)  # checks DB too
         if original is None:
             return None
@@ -318,6 +356,7 @@ class SessionManager:
 
     def update_cwd(self, session_id: str, cwd: str) -> Optional[SessionState]:
         """Update the working directory for a session and its tool overrides."""
+        cwd = _translate_acp_cwd(cwd)
         state = self.get_session(session_id)  # checks DB too
         if state is None:
             return None
@@ -427,17 +466,10 @@ class SessionManager:
                 except Exception:
                     logger.debug("Failed to update ACP session metadata", exc_info=True)
 
-            # Replace stored messages with current history.
-            db.clear_messages(state.session_id)
-            for msg in state.history:
-                db.append_message(
-                    session_id=state.session_id,
-                    role=msg.get("role", "user"),
-                    content=msg.get("content"),
-                    tool_name=msg.get("tool_name") or msg.get("name"),
-                    tool_calls=msg.get("tool_calls"),
-                    tool_call_id=msg.get("tool_call_id"),
-                )
+            # Replace stored messages with current history atomically so a
+            # mid-rewrite failure rolls back and the previously persisted
+            # conversation is preserved (salvaged from #13675).
+            db.replace_messages(state.session_id, state.history)
         except Exception:
             logger.warning("Failed to persist ACP session %s", state.session_id, exc_info=True)
 

acp_adapter/tools.py

@@ -28,6 +28,11 @@ TOOL_KIND_MAP: Dict[str, ToolKind] = {
     "terminal": "execute",
     "process": "execute",
     "execute_code": "execute",
+    # Session/meta tools
+    "todo": "other",
+    "skill_view": "read",
+    "skills_list": "read",
+    "skill_manage": "edit",
     # Web / fetch
     "web_search": "fetch",
     "web_extract": "fetch",
@@ -51,6 +56,28 @@ TOOL_KIND_MAP: Dict[str, ToolKind] = {
 }
 
 
+_POLISHED_TOOLS = {
+    # Core operator loop
+    "todo", "memory", "session_search", "delegate_task",
+    # Files / execution
+    "read_file", "write_file", "patch", "search_files", "terminal", "process", "execute_code",
+    # Skills / web / browser / media
+    "skill_view", "skills_list", "skill_manage", "web_search", "web_extract",
+    "browser_navigate", "browser_click", "browser_type", "browser_press", "browser_scroll",
+    "browser_back", "browser_snapshot", "browser_console", "browser_get_images", "browser_vision",
+    "vision_analyze", "image_generate", "text_to_speech",
+    # Schedulers / platform integrations
+    "cronjob", "send_message", "clarify", "discord", "discord_admin",
+    "ha_list_entities", "ha_get_state", "ha_list_services", "ha_call_service",
+    "feishu_doc_read", "feishu_drive_list_comments", "feishu_drive_list_comment_replies",
+    "feishu_drive_reply_comment", "feishu_drive_add_comment",
+    "kanban_create", "kanban_show", "kanban_comment", "kanban_complete",
+    "kanban_block", "kanban_link", "kanban_heartbeat",
+    "yb_query_group_info", "yb_query_group_members", "yb_search_sticker",
+    "yb_send_dm", "yb_send_sticker", "mixture_of_agents",
+}
+
+
 def get_tool_kind(tool_name: str) -> ToolKind:
     """Return the ACP ToolKind for a hermes tool, defaulting to 'other'."""
     return TOOL_KIND_MAP.get(tool_name, "other")
@@ -85,18 +112,645 @@ def build_tool_title(tool_name: str, args: Dict[str, Any]) -> str:
         if urls:
             return f"extract: {urls[0]}" + (f" (+{len(urls)-1})" if len(urls) > 1 else "")
         return "web extract"
+    if tool_name == "process":
+        action = str(args.get("action") or "").strip() or "manage"
+        sid = str(args.get("session_id") or "").strip()
+        return f"process {action}: {sid}" if sid else f"process {action}"
     if tool_name == "delegate_task":
+        tasks = args.get("tasks")
+        if isinstance(tasks, list) and tasks:
+            return f"delegate batch ({len(tasks)} tasks)"
         goal = args.get("goal", "")
         if goal and len(goal) > 60:
             goal = goal[:57] + "..."
         return f"delegate: {goal}" if goal else "delegate task"
+    if tool_name == "session_search":
+        query = str(args.get("query") or "").strip()
+        return f"session search: {query}" if query else "recent sessions"
+    if tool_name == "memory":
+        action = str(args.get("action") or "manage").strip() or "manage"
+        target = str(args.get("target") or "memory").strip() or "memory"
+        return f"memory {action}: {target}"
     if tool_name == "execute_code":
-        return "execute code"
+        code = str(args.get("code") or "").strip()
+        first_line = next((line.strip() for line in code.splitlines() if line.strip()), "")
+        if first_line:
+            if len(first_line) > 70:
+                first_line = first_line[:67] + "..."
+            return f"python: {first_line}"
+        return "python code"
+    if tool_name == "todo":
+        items = args.get("todos")
+        if isinstance(items, list):
+            return f"todo ({len(items)} item{'s' if len(items) != 1 else ''})"
+        return "todo"
+    if tool_name == "skill_view":
+        name = str(args.get("name") or "?").strip() or "?"
+        file_path = str(args.get("file_path") or "").strip()
+        suffix = f"/{file_path}" if file_path else ""
+        return f"skill view ({name}{suffix})"
+    if tool_name == "skills_list":
+        category = str(args.get("category") or "").strip()
+        return f"skills list ({category})" if category else "skills list"
+    if tool_name == "skill_manage":
+        action = str(args.get("action") or "manage").strip() or "manage"
+        name = str(args.get("name") or "?").strip() or "?"
+        file_path = str(args.get("file_path") or "").strip()
+        target = f"{name}/{file_path}" if file_path else name
+        if len(target) > 64:
+            target = target[:61] + "..."
+        return f"skill {action}: {target}"
+    if tool_name == "browser_navigate":
+        return f"navigate: {args.get('url', '?')}"
+    if tool_name == "browser_snapshot":
+        return "browser snapshot"
+    if tool_name == "browser_vision":
+        return f"browser vision: {str(args.get('question', '?'))[:50]}"
+    if tool_name == "browser_get_images":
+        return "browser images"
     if tool_name == "vision_analyze":
-        return f"analyze image: {args.get('question', '?')[:50]}"
+        return f"analyze image: {str(args.get('question', '?'))[:50]}"
+    if tool_name == "image_generate":
+        prompt = str(args.get("prompt") or args.get("description") or "").strip()
+        return f"generate image: {prompt[:50]}" if prompt else "generate image"
+    if tool_name == "cronjob":
+        action = str(args.get("action") or "manage").strip() or "manage"
+        job_id = str(args.get("job_id") or args.get("id") or "").strip()
+        return f"cron {action}: {job_id}" if job_id else f"cron {action}"
     return tool_name
 
 
+def _text(content: str) -> Any:
+    return acp.tool_content(acp.text_block(content))
+
+
+def _json_loads_maybe(value: Optional[str]) -> Any:
+    if not isinstance(value, str):
+        return value
+    try:
+        return json.loads(value)
+    except Exception:
+        pass
+
+    # Some Hermes tools append a human hint after a JSON payload, e.g.
+    # ``{...}\n\n[Hint: Results truncated...]``. Keep the structured rendering path
+    # by decoding the first JSON value instead of falling back to raw text.
+    try:
+        decoded, _ = json.JSONDecoder().raw_decode(value.lstrip())
+        return decoded
+    except Exception:
+        return None
+
+
+def _truncate_text(text: str, limit: int = 5000) -> str:
+    if len(text) <= limit:
+        return text
+    return text[: max(0, limit - 100)] + f"\n... ({len(text)} chars total, truncated)"
+
+
+def _fenced_text(text: str, language: str = "") -> str:
+    """Return a Markdown fence that cannot be broken by backticks in text."""
+    longest = max((len(run) for run in text.split("`")[1::2]), default=0)
+    fence = "`" * max(3, longest + 1)
+    return f"{fence}{language}\n{text}\n{fence}"
+
+
+def _format_todo_result(result: Optional[str]) -> Optional[str]:
+    data = _json_loads_maybe(result)
+    if not isinstance(data, dict) or not isinstance(data.get("todos"), list):
+        return None
+    summary = data.get("summary") if isinstance(data.get("summary"), dict) else {}
+    icon = {
+        "completed": "✅",
+        "in_progress": "🔄",
+        "pending": "⏳",
+        "cancelled": "✗",
+    }
+    lines = ["**Todo list**", ""]
+    for item in data["todos"]:
+        if not isinstance(item, dict):
+            continue
+        status = str(item.get("status") or "pending")
+        content = str(item.get("content") or item.get("id") or "").strip()
+        if content:
+            lines.append(f"- {icon.get(status, '•')} {content}")
+    if summary:
+        cancelled = summary.get("cancelled", 0)
+        lines.extend([
+            "",
+            "**Progress:** "
+            f"{summary.get('completed', 0)} completed, "
+            f"{summary.get('in_progress', 0)} in progress, "
+            f"{summary.get('pending', 0)} pending"
+            + (f", {cancelled} cancelled" if cancelled else ""),
+        ])
+    return "\n".join(lines)
+
+
+def _format_read_file_result(result: Optional[str], args: Optional[Dict[str, Any]]) -> Optional[str]:
+    data = _json_loads_maybe(result)
+    if not isinstance(data, dict):
+        return None
+    if data.get("error") and not data.get("content"):
+        return f"Read failed: {data.get('error')}"
+    content = data.get("content")
+    if not isinstance(content, str):
+        return None
+    path = str((args or {}).get("path") or data.get("path") or "file").strip()
+    offset = (args or {}).get("offset")
+    limit = (args or {}).get("limit")
+    range_bits = []
+    if offset:
+        range_bits.append(f"from line {offset}")
+    if limit:
+        range_bits.append(f"limit {limit}")
+    suffix = f" ({', '.join(range_bits)})" if range_bits else ""
+    header = f"Read {path}{suffix}"
+    if data.get("total_lines") is not None:
+        header += f" — {data.get('total_lines')} total lines"
+    # Hermes read_file output is line-numbered with `|`. If we send it as raw
+    # Markdown, Zed can interpret pipes as tables and collapse the layout.
+    # Fence the payload so file lines stay readable and literal.
+    return _truncate_text(f"{header}\n\n{_fenced_text(content)}")
+
+
+def _format_search_files_result(result: Optional[str]) -> Optional[str]:
+    data = _json_loads_maybe(result)
+    if not isinstance(data, dict):
+        return None
+    matches = data.get("matches")
+    if not isinstance(matches, list):
+        return None
+
+    total = data.get("total_count", len(matches))
+    shown = min(len(matches), 12)
+    truncated = bool(data.get("truncated")) or len(matches) > shown
+    lines = [
+        "Search results",
+        f"Found {total} match{'es' if total != 1 else ''}; showing {shown}.",
+        "",
+    ]
+
+    for match in matches[:shown]:
+        if not isinstance(match, dict):
+            lines.append(f"- {match}")
+            continue
+
+        path = str(match.get("path") or match.get("file") or match.get("filename") or "?")
+        line = match.get("line") or match.get("line_number")
+        content = str(match.get("content") or match.get("text") or "").strip()
+        loc = f"{path}:{line}" if line else path
+        lines.append(f"- {loc}")
+        if content:
+            snippet = _truncate_text(" ".join(content.split()), 300)
+            lines.append(f"  {snippet}")
+
+    if truncated:
+        lines.extend([
+            "",
+            "Results truncated. Narrow the search, add file_glob, or use offset to page.",
+        ])
+    return _truncate_text("\n".join(lines), limit=7000)
+
+
+def _format_execute_code_result(result: Optional[str]) -> Optional[str]:
+    data = _json_loads_maybe(result)
+    if not isinstance(data, dict):
+        return result if isinstance(result, str) and result.strip() else None
+    output = str(data.get("output") or "")
+    error = str(data.get("error") or "")
+    exit_code = data.get("exit_code")
+    parts = [f"Exit code: {exit_code}" if exit_code is not None else "Execution complete"]
+    if output:
+        parts.extend(["", "Output:", output])
+    if error:
+        parts.extend(["", "Error:", error])
+    return _truncate_text("\n".join(parts))
+
+
+def _extract_markdown_headings(content: str, limit: int = 8) -> list[str]:
+    headings: list[str] = []
+    for line in content.splitlines():
+        stripped = line.strip()
+        if stripped.startswith("#"):
+            heading = stripped.lstrip("#").strip()
+            if heading:
+                headings.append(heading)
+        if len(headings) >= limit:
+            break
+    return headings
+
+
+def _format_skill_view_result(result: Optional[str]) -> Optional[str]:
+    data = _json_loads_maybe(result)
+    if not isinstance(data, dict):
+        return None
+    if data.get("success") is False:
+        return f"Skill view failed: {data.get('error', 'unknown error')}"
+    name = str(data.get("name") or "skill")
+    file_path = str(data.get("file") or data.get("path") or "SKILL.md")
+    description = str(data.get("description") or "").strip()
+    content = str(data.get("content") or "")
+    linked = data.get("linked_files") if isinstance(data.get("linked_files"), dict) else None
+
+    lines = ["**Skill loaded**", "", f"- **Name:** `{name}`", f"- **File:** `{file_path}`"]
+    if description:
+        lines.append(f"- **Description:** {description}")
+    if content:
+        lines.append(f"- **Content:** {len(content):,} chars loaded into agent context")
+    if linked:
+        linked_count = sum(len(v) for v in linked.values() if isinstance(v, list))
+        lines.append(f"- **Linked files:** {linked_count}")
+
+    headings = _extract_markdown_headings(content)
+    if headings:
+        lines.extend(["", "**Sections**"])
+        lines.extend(f"- {heading}" for heading in headings)
+
+    lines.extend([
+        "",
+        "_Full skill content is available to the agent but hidden here to keep ACP readable._",
+    ])
+    return "\n".join(lines)
+
+
+def _format_skill_manage_result(result: Optional[str], args: Optional[Dict[str, Any]]) -> Optional[str]:
+    data = _json_loads_maybe(result)
+    if not isinstance(data, dict):
+        return None
+
+    action = str((args or {}).get("action") or "manage").strip() or "manage"
+    name = str((args or {}).get("name") or data.get("name") or "skill").strip() or "skill"
+    file_path = str((args or {}).get("file_path") or data.get("file_path") or "SKILL.md").strip() or "SKILL.md"
+    success = data.get("success")
+    status = "✅ Skill updated" if success is not False else "✗ Skill update failed"
+
+    lines = [f"**{status}**", "", f"- **Action:** `{action}`", f"- **Skill:** `{name}`"]
+    if action not in {"delete"}:
+        lines.append(f"- **File:** `{file_path}`")
+
+    message = str(data.get("message") or data.get("error") or "").strip()
+    if message:
+        lines.append(f"- **Result:** {message}")
+
+    replacements = data.get("replacements") or data.get("replacement_count")
+    if replacements is not None:
+        lines.append(f"- **Replacements:** {replacements}")
+
+    path = str(data.get("path") or "").strip()
+    if path:
+        lines.append(f"- **Path:** `{path}`")
+
+    return "\n".join(lines)
+
+
+def _format_web_search_result(result: Optional[str]) -> Optional[str]:
+    data = _json_loads_maybe(result)
+    if not isinstance(data, dict):
+        return None
+    web = data.get("data", {}).get("web") if isinstance(data.get("data"), dict) else data.get("web")
+    if not isinstance(web, list):
+        return None
+    lines = [f"Web results: {len(web)}"]
+    for item in web[:10]:
+        if not isinstance(item, dict):
+            continue
+        title = str(item.get("title") or item.get("url") or "result").strip()
+        url = str(item.get("url") or "").strip()
+        desc = str(item.get("description") or "").strip()
+        lines.append(f"• {title}" + (f" — {url}" if url else ""))
+        if desc:
+            lines.append(f"  {desc}")
+    return _truncate_text("\n".join(lines))
+
+
+def _format_web_extract_result(result: Optional[str]) -> Optional[str]:
+    """Return only web_extract errors for ACP; success stays compact via title."""
+    data = _json_loads_maybe(result)
+    if not isinstance(data, dict):
+        return None
+    if data.get("success") is False and data.get("error"):
+        return f"Web extract failed: {data.get('error')}"
+    results = data.get("results")
+    if not isinstance(results, list):
+        return None
+
+    failures: list[str] = []
+    for item in results[:10]:
+        if not isinstance(item, dict):
+            continue
+        error = str(item.get("error") or "").strip()
+        if not error or error in {"None", "null"}:
+            continue
+        url = str(item.get("url") or "").strip()
+        title = str(item.get("title") or url or "Untitled").strip()
+        failures.append(
+            f"- {title}" + (f" — {url}" if url and url != title else "") + f"\n  Error: {_truncate_text(error, limit=500)}"
+        )
+
+    if not failures:
+        return None
+    lines = [f"Web extract failed for {len(failures)} URL{'s' if len(failures) != 1 else ''}"]
+    lines.extend(failures)
+    return "\n".join(lines)
+
+
+def _format_process_result(result: Optional[str], args: Optional[Dict[str, Any]]) -> Optional[str]:
+    data = _json_loads_maybe(result)
+    if not isinstance(data, dict):
+        return result if isinstance(result, str) and result.strip() else None
+    if data.get("success") is False and data.get("error"):
+        return f"Process error: {data.get('error')}"
+    action = str((args or {}).get("action") or "process").strip() or "process"
+    if isinstance(data.get("processes"), list):
+        processes = data["processes"]
+        lines = [f"Processes: {len(processes)}"]
+        for proc in processes[:20]:
+            if not isinstance(proc, dict):
+                lines.append(f"- {proc}")
+                continue
+            sid = str(proc.get("session_id") or proc.get("id") or "?")
+            status = str(proc.get("status") or ("exited" if proc.get("exited") else "running"))
+            cmd = str(proc.get("command") or "").strip()
+            pid = proc.get("pid")
+            code = proc.get("exit_code")
+            bits = [status]
+            if pid is not None:
+                bits.append(f"pid {pid}")
+            if code is not None:
+                bits.append(f"exit {code}")
+            lines.append(f"- `{sid}` — {', '.join(bits)}" + (f" — {cmd[:120]}" if cmd else ""))
+        if len(processes) > 20:
+            lines.append(f"... {len(processes) - 20} more process(es)")
+        return "\n".join(lines)
+
+    status = str(data.get("status") or data.get("state") or action).strip()
+    sid = str(data.get("session_id") or (args or {}).get("session_id") or "").strip()
+    lines = [f"Process {action}: {status}" + (f" (`{sid}`)" if sid else "")]
+    for key, label in (("command", "Command"), ("pid", "PID"), ("exit_code", "Exit code"), ("returncode", "Exit code"), ("lines", "Lines")):
+        if data.get(key) is not None:
+            lines.append(f"- **{label}:** {data.get(key)}")
+    output = data.get("output") or data.get("new_output") or data.get("log") or data.get("stdout")
+    error = data.get("error") or data.get("stderr")
+    if output:
+        lines.extend(["", "Output:", _truncate_text(str(output), limit=5000)])
+    if error:
+        lines.extend(["", "Error:", _truncate_text(str(error), limit=2000)])
+    msg = data.get("message")
+    if msg and not output and not error:
+        lines.append(str(msg))
+    return _truncate_text("\n".join(lines), limit=7000)
+
+
+def _format_delegate_result(result: Optional[str]) -> Optional[str]:
+    data = _json_loads_maybe(result)
+    if not isinstance(data, dict):
+        return None
+    if data.get("error") and not isinstance(data.get("results"), list):
+        return f"Delegation failed: {data.get('error')}"
+    results = data.get("results")
+    if not isinstance(results, list):
+        return None
+    total = data.get("total_duration_seconds")
+    lines = [f"Delegation results: {len(results)} task{'s' if len(results) != 1 else ''}" + (f" in {total}s" if total is not None else "")]
+    icon = {"completed": "✅", "failed": "✗", "error": "✗", "timeout": "⏱", "interrupted": "⚠"}
+    for item in results:
+        if not isinstance(item, dict):
+            lines.append(f"- {item}")
+            continue
+        idx = item.get("task_index")
+        status = str(item.get("status") or "unknown")
+        model = item.get("model")
+        dur = item.get("duration_seconds")
+        role = item.get("_child_role")
+        header = f"{icon.get(status, '•')} Task {idx + 1 if isinstance(idx, int) else '?'}: {status}"
+        bits = []
+        if model:
+            bits.append(str(model))
+        if role:
+            bits.append(f"role={role}")
+        if dur is not None:
+            bits.append(f"{dur}s")
+        if bits:
+            header += " (" + ", ".join(bits) + ")"
+        lines.extend(["", header])
+        summary = str(item.get("summary") or "").strip()
+        error = str(item.get("error") or "").strip()
+        if summary:
+            lines.append(_truncate_text(summary, limit=1200))
+        if error:
+            lines.append("Error: " + _truncate_text(error, limit=800))
+        trace = item.get("tool_trace")
+        if isinstance(trace, list) and trace:
+            names = [str(t.get("tool") or "?") for t in trace if isinstance(t, dict)]
+            if names:
+                lines.append("Tools: " + ", ".join(names[:12]) + (f" (+{len(names)-12})" if len(names) > 12 else ""))
+    return _truncate_text("\n".join(lines), limit=8000)
+
+
+def _format_session_search_result(result: Optional[str]) -> Optional[str]:
+    data = _json_loads_maybe(result)
+    if not isinstance(data, dict):
+        return None
+    if data.get("success") is False:
+        return f"Session search failed: {data.get('error', 'unknown error')}"
+    results = data.get("results")
+    if not isinstance(results, list):
+        return None
+    mode = data.get("mode") or "search"
+    query = data.get("query")
+    lines = ["Recent sessions" if mode == "recent" else f"Session search results" + (f" for `{query}`" if query else "")]
+    if not results:
+        lines.append(str(data.get("message") or "No matching sessions found."))
+        return "\n".join(lines)
+    for item in results:
+        if not isinstance(item, dict):
+            continue
+        sid = str(item.get("session_id") or "?")
+        title = str(item.get("title") or item.get("when") or "Untitled session").strip()
+        when = str(item.get("last_active") or item.get("started_at") or item.get("when") or "").strip()
+        count = item.get("message_count")
+        source = str(item.get("source") or "").strip()
+        meta = ", ".join(str(x) for x in [when, source, f"{count} msgs" if count is not None else ""] if x)
+        lines.append(f"- **{title}** (`{sid}`)" + (f" — {meta}" if meta else ""))
+        summary = str(item.get("summary") or item.get("preview") or "").strip()
+        if summary:
+            lines.append("  " + _truncate_text(" ".join(summary.split()), limit=500))
+    return _truncate_text("\n".join(lines), limit=7000)
+
+
+def _format_memory_result(result: Optional[str], args: Optional[Dict[str, Any]]) -> Optional[str]:
+    data = _json_loads_maybe(result)
+    if not isinstance(data, dict):
+        return None
+    action = str((args or {}).get("action") or "memory").strip() or "memory"
+    target = str(data.get("target") or (args or {}).get("target") or "memory")
+    if data.get("success") is False:
+        lines = [f"✗ Memory {action} failed ({target})", str(data.get("error") or "unknown error")]
+        matches = data.get("matches")
+        if isinstance(matches, list) and matches:
+            lines.append("Matches:")
+            lines.extend(f"- {_truncate_text(str(m), 160)}" for m in matches[:5])
+        return "\n".join(lines)
+    lines = [f"✅ Memory {action} saved ({target})"]
+    if data.get("message"):
+        lines.append(str(data.get("message")))
+    if data.get("entry_count") is not None:
+        lines.append(f"Entries: {data.get('entry_count')}")
+    if data.get("usage"):
+        lines.append(f"Usage: {data.get('usage')}")
+    # Avoid dumping all memory entries into ACP UI; show only the explicit new value preview.
+    preview = str((args or {}).get("content") or (args or {}).get("old_text") or "").strip()
+    if preview:
+        lines.append("Preview: " + _truncate_text(preview, limit=300))
+    return "\n".join(lines)
+
+
+def _format_edit_result(tool_name: str, result: Optional[str], args: Optional[Dict[str, Any]]) -> Optional[str]:
+    data = _json_loads_maybe(result)
+    path = str((args or {}).get("path") or "file").strip()
+    if isinstance(data, dict):
+        if data.get("success") is False or data.get("error"):
+            return f"{tool_name} failed for {path}: {data.get('error', 'unknown error')}"
+        message = str(data.get("message") or "").strip()
+        replacements = data.get("replacements") or data.get("replacement_count")
+        lines = [f"✅ {tool_name} completed" + (f" for `{path}`" if path else "")]
+        if message:
+            lines.append(message)
+        if replacements is not None:
+            lines.append(f"Replacements: {replacements}")
+        if data.get("files_modified"):
+            files = data.get("files_modified")
+            if isinstance(files, list):
+                lines.append("Files: " + ", ".join(f"`{f}`" for f in files[:8]))
+        return "\n".join(lines)
+    if isinstance(result, str) and result.strip():
+        return _truncate_text(result, limit=3000)
+    return f"✅ {tool_name} completed" + (f" for `{path}`" if path else "")
+
+
+def _format_browser_result(tool_name: str, result: Optional[str], args: Optional[Dict[str, Any]]) -> Optional[str]:
+    data = _json_loads_maybe(result)
+    if not isinstance(data, dict):
+        return result if isinstance(result, str) and result.strip() else None
+    if data.get("success") is False or data.get("error"):
+        return f"{tool_name} failed: {data.get('error', 'unknown error')}"
+    if tool_name == "browser_get_images":
+        images = data.get("images") or data.get("data")
+        if isinstance(images, list):
+            lines = [f"Images found: {len(images)}"]
+            for img in images[:12]:
+                if isinstance(img, dict):
+                    alt = str(img.get("alt") or "").strip()
+                    url = str(img.get("url") or img.get("src") or "").strip()
+                    lines.append(f"- {alt or 'image'}" + (f" — {url}" if url else ""))
+            return _truncate_text("\n".join(lines), limit=5000)
+    title = str(data.get("title") or data.get("url") or data.get("status") or tool_name)
+    text = str(data.get("text") or data.get("content") or data.get("snapshot") or data.get("analysis") or data.get("message") or "").strip()
+    lines = [title]
+    if data.get("url") and data.get("url") != title:
+        lines.append(str(data.get("url")))
+    if text:
+        lines.extend(["", _truncate_text(text, limit=5000)])
+    return _truncate_text("\n".join(lines), limit=7000)
+
+
+def _format_media_or_cron_result(tool_name: str, result: Optional[str]) -> Optional[str]:
+    data = _json_loads_maybe(result)
+    if not isinstance(data, dict):
+        return result if isinstance(result, str) and result.strip() else None
+    if data.get("success") is False or data.get("error"):
+        return f"{tool_name} failed: {data.get('error', 'unknown error')}"
+    lines = [f"✅ {tool_name} completed"]
+    for key in ("file_path", "path", "url", "image_url", "job_id", "id", "status", "message", "next_run"):
+        if data.get(key):
+            lines.append(f"- **{key}:** {data.get(key)}")
+    return "\n".join(lines)
+
+
+def _format_generic_structured_result(tool_name: str, result: Optional[str]) -> Optional[str]:
+    data = _json_loads_maybe(result)
+    if not isinstance(data, (dict, list)):
+        return result if isinstance(result, str) and result.strip() else None
+    if isinstance(data, list):
+        lines = [f"{tool_name}: {len(data)} item{'s' if len(data) != 1 else ''}"]
+        for item in data[:12]:
+            lines.append(f"- {_truncate_text(str(item), limit=240)}")
+        return _truncate_text("\n".join(lines), limit=5000)
+
+    if data.get("success") is False or data.get("error"):
+        return f"{tool_name} failed: {data.get('error', 'unknown error')}"
+
+    lines = [f"✅ {tool_name} completed" if data.get("success") is True else f"{tool_name} result"]
+    priority_keys = (
+        "message", "status", "id", "task_id", "issue_id", "title", "name", "entity_id",
+        "state", "service", "url", "path", "file_path", "count", "total", "next_run",
+    )
+    seen = set()
+    for key in priority_keys:
+        value = data.get(key)
+        if value in (None, "", [], {}):
+            continue
+        seen.add(key)
+        lines.append(f"- **{key}:** {_truncate_text(str(value), limit=500)}")
+
+    for key, value in data.items():
+        if key in seen or key in {"success", "raw", "content", "entries"}:
+            continue
+        if value in (None, "", [], {}):
+            continue
+        if isinstance(value, (dict, list)):
+            preview = json.dumps(value, ensure_ascii=False, default=str)
+        else:
+            preview = str(value)
+        lines.append(f"- **{key}:** {_truncate_text(preview, limit=500)}")
+        if len(lines) >= 14:
+            break
+
+    content = data.get("content")
+    if isinstance(content, str) and content.strip():
+        lines.extend(["", _truncate_text(content.strip(), limit=1500)])
+    return _truncate_text("\n".join(lines), limit=7000)
+
+
+def _build_polished_completion_content(
+    tool_name: str,
+    result: Optional[str],
+    function_args: Optional[Dict[str, Any]],
+) -> Optional[List[Any]]:
+    formatter = {
+        "todo": lambda: _format_todo_result(result),
+        "read_file": lambda: _format_read_file_result(result, function_args),
+        "write_file": lambda: _format_edit_result(tool_name, result, function_args),
+        "patch": lambda: _format_edit_result(tool_name, result, function_args),
+        "search_files": lambda: _format_search_files_result(result),
+        "execute_code": lambda: _format_execute_code_result(result),
+        "process": lambda: _format_process_result(result, function_args),
+        "delegate_task": lambda: _format_delegate_result(result),
+        "session_search": lambda: _format_session_search_result(result),
+        "memory": lambda: _format_memory_result(result, function_args),
+        "skill_view": lambda: _format_skill_view_result(result),
+        "skill_manage": lambda: _format_skill_manage_result(result, function_args),
+        "web_search": lambda: _format_web_search_result(result),
+        "web_extract": lambda: _format_web_extract_result(result),
+        "browser_navigate": lambda: _format_browser_result(tool_name, result, function_args),
+        "browser_snapshot": lambda: _format_browser_result(tool_name, result, function_args),
+        "browser_vision": lambda: _format_browser_result(tool_name, result, function_args),
+        "browser_get_images": lambda: _format_browser_result(tool_name, result, function_args),
+        "vision_analyze": lambda: _format_media_or_cron_result(tool_name, result),
+        "image_generate": lambda: _format_media_or_cron_result(tool_name, result),
+        "cronjob": lambda: _format_media_or_cron_result(tool_name, result),
+    }.get(tool_name)
+    if formatter is None and tool_name in _POLISHED_TOOLS:
+        formatter = lambda: _format_generic_structured_result(tool_name, result)
+    if formatter is None:
+        return None
+    text = formatter()
+    if not text:
+        return None
+    return [_text(text)]
+
+
 def _build_patch_mode_content(patch_text: str) -> List[Any]:
     """Parse V4A patch mode input into ACP diff blocks when possible."""
     if not patch_text:
@@ -258,7 +912,11 @@ def _build_tool_complete_content(
         except Exception:
             pass
 
-    return [acp.tool_content(acp.text_block(display_result))]
+    polished_content = _build_polished_completion_content(tool_name, result, function_args)
+    if polished_content:
+        return polished_content
+
+    return [_text(display_result)]
 
 
 # ---------------------------------------------------------------------------
@@ -288,7 +946,6 @@ def build_tool_start(
             content = _build_patch_mode_content(patch_text)
         return acp.start_tool_call(
             tool_call_id, title, kind=kind, content=content, locations=locations,
-            raw_input=arguments,
         )
 
     if tool_name == "write_file":
@@ -297,32 +954,172 @@ def build_tool_start(
         content = [acp.tool_diff_content(path=path, new_text=file_content)]
         return acp.start_tool_call(
             tool_call_id, title, kind=kind, content=content, locations=locations,
-            raw_input=arguments,
         )
 
     if tool_name == "terminal":
         command = arguments.get("command", "")
-        content = [acp.tool_content(acp.text_block(f"$ {command}"))]
+        content = [_text(f"$ {command}")]
         return acp.start_tool_call(
             tool_call_id, title, kind=kind, content=content, locations=locations,
-            raw_input=arguments,
         )
 
     if tool_name == "read_file":
-        path = arguments.get("path", "")
-        content = [acp.tool_content(acp.text_block(f"Reading {path}"))]
+        # The title and location already identify the file. Sending a synthetic
+        # "Reading ..." content block makes Zed render an unhelpful Output
+        # section before the real file contents arrive on completion.
         return acp.start_tool_call(
-            tool_call_id, title, kind=kind, content=content, locations=locations,
-            raw_input=arguments,
+            tool_call_id, title, kind=kind, content=None, locations=locations,
         )
 
     if tool_name == "search_files":
         pattern = arguments.get("pattern", "")
         target = arguments.get("target", "content")
-        content = [acp.tool_content(acp.text_block(f"Searching for '{pattern}' ({target})"))]
+        search_path = arguments.get("path")
+        where = f" in {search_path}" if search_path else ""
+        content = [_text(f"Searching for '{pattern}' ({target}){where}")]
+        return acp.start_tool_call(
+            tool_call_id, title, kind=kind, content=content, locations=locations,
+        )
+
+    if tool_name == "todo":
+        items = arguments.get("todos")
+        if isinstance(items, list):
+            preview_lines = ["Updating todo list", ""]
+            for item in items[:8]:
+                if isinstance(item, dict):
+                    preview_lines.append(f"- {item.get('status', 'pending')}: {item.get('content', item.get('id', ''))}")
+            if len(items) > 8:
+                preview_lines.append(f"... {len(items) - 8} more")
+            content = [_text("\n".join(preview_lines))]
+        else:
+            content = [_text("Reading todo list")]
+        return acp.start_tool_call(
+            tool_call_id, title, kind=kind, content=content, locations=locations,
+        )
+
+    if tool_name == "skill_view":
+        name = str(arguments.get("name") or "?").strip() or "?"
+        file_path = str(arguments.get("file_path") or "SKILL.md").strip() or "SKILL.md"
+        content = [_text(f"Loading skill '{name}' ({file_path})")]
+        return acp.start_tool_call(
+            tool_call_id, title, kind=kind, content=content, locations=locations,
+        )
+
+    if tool_name == "skill_manage":
+        action = str(arguments.get("action") or "manage").strip() or "manage"
+        name = str(arguments.get("name") or "?").strip() or "?"
+        file_path = str(arguments.get("file_path") or "SKILL.md").strip() or "SKILL.md"
+        path = f"skills/{name}/{file_path}" if file_path else f"skills/{name}"
+
+        if action == "patch":
+            old = str(arguments.get("old_string") or "")
+            new = str(arguments.get("new_string") or "")
+            content = [acp.tool_diff_content(path=path, old_text=old or None, new_text=new)]
+        elif action in {"edit", "create"}:
+            content = [
+                acp.tool_diff_content(
+                    path=path,
+                    new_text=str(arguments.get("content") or ""),
+                )
+            ]
+        elif action == "write_file":
+            target = str(arguments.get("file_path") or "file")
+            content = [
+                acp.tool_diff_content(
+                    path=f"skills/{name}/{target}",
+                    new_text=str(arguments.get("file_content") or ""),
+                )
+            ]
+        elif action in {"delete", "remove_file"}:
+            target = str(arguments.get("file_path") or file_path or name)
+            content = [_text(f"Removing {target} from skill '{name}'")]
+        else:
+            content = [_text(f"Running skill_manage action '{action}' on skill '{name}' ({file_path})")]
+
+        return acp.start_tool_call(
+            tool_call_id, title, kind=kind, content=content, locations=locations,
+        )
+
+    if tool_name == "execute_code":
+        code = str(arguments.get("code") or "").strip()
+        preview = code[:1200] + (f"\n... ({len(code)} chars total, truncated)" if len(code) > 1200 else "")
+        content = [_text(f"Running Python helper script:\n\n```python\n{preview}\n```" if preview else "Running Python helper script")]
+        return acp.start_tool_call(
+            tool_call_id, title, kind=kind, content=content, locations=locations,
+        )
+
+    if tool_name == "web_search":
+        query = str(arguments.get("query") or "").strip()
+        content = [_text(f"Searching the web for: {query}" if query else "Searching the web")]
+        return acp.start_tool_call(
+            tool_call_id, title, kind=kind, content=content, locations=locations,
+        )
+
+    if tool_name == "web_extract":
+        # The title identifies the URL(s). Avoid a duplicate content block so
+        # Zed renders this like read_file: compact start, concise completion.
+        return acp.start_tool_call(
+            tool_call_id, title, kind=kind, content=None, locations=locations,
+        )
+
+    if tool_name == "process":
+        action = str(arguments.get("action") or "").strip() or "manage"
+        sid = str(arguments.get("session_id") or "").strip()
+        data_preview = str(arguments.get("data") or "").strip()
+        text = f"Process action: {action}" + (f"\nSession: {sid}" if sid else "")
+        if data_preview:
+            text += "\nInput: " + _truncate_text(data_preview, limit=500)
+        content = [_text(text)]
+        return acp.start_tool_call(
+            tool_call_id, title, kind=kind, content=content, locations=locations,
+        )
+
+    if tool_name == "delegate_task":
+        tasks = arguments.get("tasks")
+        if isinstance(tasks, list) and tasks:
+            lines = [f"Delegating {len(tasks)} tasks", ""]
+            for i, task in enumerate(tasks[:8], 1):
+                if isinstance(task, dict):
+                    goal = str(task.get("goal") or "").strip()
+                    role = str(task.get("role") or "").strip()
+                    lines.append(f"{i}. " + _truncate_text(goal, limit=160) + (f" ({role})" if role else ""))
+            if len(tasks) > 8:
+                lines.append(f"... {len(tasks) - 8} more")
+            content = [_text("\n".join(lines))]
+        else:
+            goal = str(arguments.get("goal") or "").strip()
+            content = [_text("Delegating task" + (f":\n{_truncate_text(goal, limit=800)}" if goal else ""))]
+        return acp.start_tool_call(
+            tool_call_id, title, kind=kind, content=content, locations=locations,
+        )
+
+    if tool_name == "session_search":
+        query = str(arguments.get("query") or "").strip()
+        content = [_text(f"Searching past sessions for: {query}" if query else "Loading recent sessions")]
+        return acp.start_tool_call(
+            tool_call_id, title, kind=kind, content=content, locations=locations,
+        )
+
+    if tool_name == "memory":
+        action = str(arguments.get("action") or "manage").strip() or "manage"
+        target = str(arguments.get("target") or "memory").strip() or "memory"
+        preview = str(arguments.get("content") or arguments.get("old_text") or "").strip()
+        text = f"Memory {action} ({target})"
+        if preview:
+            text += "\nPreview: " + _truncate_text(preview, limit=500)
+        content = [_text(text)]
+        return acp.start_tool_call(
+            tool_call_id, title, kind=kind, content=content, locations=locations,
+        )
+
+    if tool_name in _POLISHED_TOOLS:
+        try:
+            args_text = json.dumps(arguments, indent=2, default=str)
+        except (TypeError, ValueError):
+            args_text = str(arguments)
+        content = [_text(_truncate_text(args_text, limit=1200))]
         return acp.start_tool_call(
             tool_call_id, title, kind=kind, content=content, locations=locations,
-            raw_input=arguments,
         )
 
     # Generic fallback
@@ -334,7 +1131,7 @@ def build_tool_start(
     content = [acp.tool_content(acp.text_block(args_text))]
     return acp.start_tool_call(
         tool_call_id, title, kind=kind, content=content, locations=locations,
-        raw_input=arguments,
+        raw_input=None if tool_name in _POLISHED_TOOLS else arguments,
     )
 
 
@@ -347,18 +1144,22 @@ def build_tool_complete(
 ) -> ToolCallProgress:
     """Create a ToolCallUpdate (progress) event for a completed tool call."""
     kind = get_tool_kind(tool_name)
-    content = _build_tool_complete_content(
-        tool_name,
-        result,
-        function_args=function_args,
-        snapshot=snapshot,
-    )
+    if tool_name == "web_extract":
+        error_text = _format_web_extract_result(result)
+        content = [_text(error_text)] if error_text else None
+    else:
+        content = _build_tool_complete_content(
+            tool_name,
+            result,
+            function_args=function_args,
+            snapshot=snapshot,
+        )
     return acp.update_tool_call(
         tool_call_id,
         kind=kind,
         status="completed",
         content=content,
-        raw_output=result,
+        raw_output=None if tool_name in _POLISHED_TOOLS else result,
     )
 
 

agent/anthropic_adapter.py

@@ -76,6 +76,7 @@ _ADAPTIVE_THINKING_SUBSTRINGS = ("4-6", "4.6", "4-7", "4.7")
 # Models where temperature/top_p/top_k return 400 if set to non-default values.
 # This is the Opus 4.7 contract; future 4.x+ models are expected to follow it.
 _NO_SAMPLING_PARAMS_SUBSTRINGS = ("4-7", "4.7")
+_FAST_MODE_SUPPORTED_SUBSTRINGS = ("opus-4-6", "opus-4.6")
 
 # ── Max output token limits per Anthropic model ───────────────────────
 # Source: Anthropic docs + Cline model catalog.  Anthropic's API requires
@@ -105,6 +106,9 @@ _ANTHROPIC_OUTPUT_LIMITS = {
     "claude-3-haiku":      4_096,
     # Third-party Anthropic-compatible providers
     "minimax":            131_072,
+    # Qwen models via DashScope Anthropic-compatible endpoint
+    # DashScope enforces max_tokens ∈ [1, 65536]
+    "qwen3":               65_536,
 }
 
 # For any model not in the table, assume the highest current limit.
@@ -216,6 +220,17 @@ def _forbids_sampling_params(model: str) -> bool:
     return any(v in model for v in _NO_SAMPLING_PARAMS_SUBSTRINGS)
 
 
+def _supports_fast_mode(model: str) -> bool:
+    """Return True for models that support Anthropic Fast Mode (speed=fast).
+
+    Per Anthropic docs, fast mode is currently supported on Opus 4.6 only.
+    Sending ``speed: "fast"`` to any other Claude model (including Opus 4.7)
+    returns HTTP 400. This guard prevents silently 400'ing when stale config
+    or older callers leave fast mode enabled across a model upgrade.
+    """
+    return any(v in model for v in _FAST_MODE_SUPPORTED_SUBSTRINGS)
+
+
 # Beta headers for enhanced features (sent with ALL auth types).
 # As of Opus 4.7 (2026-04-16), the first two are GA on Claude 4.6+ — the
 # beta headers are still accepted (harmless no-op) but not required. Kept
@@ -1222,6 +1237,14 @@ def _normalize_tool_input_schema(schema: Any) -> Dict[str, Any]:
     ``keep_nullable_hint=False`` because the Anthropic validator does not
     recognize the OpenAPI-style ``nullable: true`` extension and strict
     schema-to-grammar converters may reject unknown keywords.
+
+    Top-level ``oneOf``/``allOf``/``anyOf`` are also stripped here: the
+    Anthropic API rejects union keywords at the schema root with a generic
+    HTTP 400. Several upstream and plugin tools ship schemas with one of
+    these keywords at the top level (commonly for Pydantic discriminated
+    unions). If we land here with those keywords still present after
+    nullable-union stripping, drop them and fall back to a plain object
+    schema so the tool still validates at the Anthropic boundary.
     """
     if not schema:
         return {"type": "object", "properties": {}}
@@ -1231,6 +1254,12 @@ def _normalize_tool_input_schema(schema: Any) -> Dict[str, Any]:
     normalized = strip_nullable_unions(schema, keep_nullable_hint=False)
     if not isinstance(normalized, dict):
         return {"type": "object", "properties": {}}
+    # Strip top-level union keywords that Anthropic's validator rejects.
+    banned = {"oneOf", "allOf", "anyOf"}
+    if banned & normalized.keys():
+        normalized = {k: v for k, v in normalized.items() if k not in banned}
+        if "type" not in normalized:
+            normalized["type"] = "object"
     if normalized.get("type") == "object" and not isinstance(normalized.get("properties"), dict):
         normalized = {**normalized, "properties": {}}
     return normalized
@@ -1241,10 +1270,24 @@ def convert_tools_to_anthropic(tools: List[Dict]) -> List[Dict]:
     if not tools:
         return []
     result = []
+    seen_names: set = set()
     for t in tools:
         fn = t.get("function", {})
+        name = fn.get("name", "")
+        # Defensive dedup: Anthropic rejects requests with duplicate tool
+        # names.  Upstream injection paths already dedup, but this guard
+        # converts a hard API failure into a warning.  See: #18478
+        if name and name in seen_names:
+            logger.warning(
+                "convert_tools_to_anthropic: duplicate tool name '%s' "
+                "— dropping second occurrence",
+                name,
+            )
+            continue
+        if name:
+            seen_names.add(name)
         result.append({
-            "name": fn.get("name", ""),
+            "name": name,
             "description": fn.get("description", ""),
             "input_schema": _normalize_tool_input_schema(
                 fn.get("parameters", {"type": "object", "properties": {}})
@@ -1901,9 +1944,15 @@ def build_anthropic_kwargs(
 
     # ── Fast mode (Opus 4.6 only) ────────────────────────────────────
     # Adds extra_body.speed="fast" + the fast-mode beta header for ~2.5x
-    # output speed. Only for native Anthropic endpoints — third-party
-    # providers would reject the unknown beta header and speed parameter.
-    if fast_mode and not _is_third_party_anthropic_endpoint(base_url):
+    # output speed. Per Anthropic docs, fast mode is only supported on
+    # Opus 4.6 — Opus 4.7 and other models 400 on the speed parameter.
+    # Only for native Anthropic endpoints — third-party providers would
+    # reject the unknown beta header and speed parameter.
+    if (
+        fast_mode
+        and not _is_third_party_anthropic_endpoint(base_url)
+        and _supports_fast_mode(model)
+    ):
         kwargs.setdefault("extra_body", {})["speed"] = "fast"
         # Build extra_headers with ALL applicable betas (the per-request
         # extra_headers override the client-level anthropic-beta header).

agent/auxiliary_client.py

@@ -259,13 +259,70 @@ _PROVIDERS_WITHOUT_VISION: frozenset = frozenset({
     "kimi-coding-cn",
 })
 
-# OpenRouter app attribution headers
-_OR_HEADERS = {
+# OpenRouter app attribution headers (base — always sent).
+# `X-Title` is the canonical attribution header OpenRouter's dashboard
+# reads; the previous `X-OpenRouter-Title` label was not recognized there.
+_OR_HEADERS_BASE = {
     "HTTP-Referer": "https://hermes-agent.nousresearch.com",
-    "X-OpenRouter-Title": "Hermes Agent",
+    "X-Title": "Hermes Agent",
     "X-OpenRouter-Categories": "productivity,cli-agent",
 }
 
+# Truthy values for boolean env-var parsing.
+_TRUTHY_ENV_VALUES = frozenset({"1", "true", "yes", "on"})
+
+
+def build_or_headers(or_config: dict | None = None) -> dict:
+    """Build OpenRouter headers, optionally including response-cache headers.
+
+    Precedence for response cache: env var > config.yaml > default (enabled).
+
+    Environment variables:
+        ``HERMES_OPENROUTER_CACHE`` — truthy (``1``/``true``/``yes``/``on``)
+            enables caching; ``0``/``false``/``no``/``off`` disables.
+            Overrides ``openrouter.response_cache`` in config.yaml.
+        ``HERMES_OPENROUTER_CACHE_TTL`` — integer seconds (1-86400).
+            Overrides ``openrouter.response_cache_ttl`` in config.yaml.
+
+    *or_config* is the ``openrouter`` section from config.yaml.  When *None*,
+    falls back to reading config from disk via ``load_config()``.
+    """
+    headers = dict(_OR_HEADERS_BASE)
+
+    # Resolve config from disk if not provided.
+    if or_config is None:
+        try:
+            from hermes_cli.config import load_config
+            or_config = load_config().get("openrouter", {})
+        except Exception:
+            or_config = {}
+
+    # Determine cache enabled: env var overrides config.
+    env_cache = os.environ.get("HERMES_OPENROUTER_CACHE", "").strip().lower()
+    if env_cache:
+        cache_enabled = env_cache in _TRUTHY_ENV_VALUES
+    else:
+        cache_enabled = or_config.get("response_cache", False)
+
+    if not cache_enabled:
+        return headers
+
+    headers["X-OpenRouter-Cache"] = "true"
+
+    # Determine TTL: env var overrides config.
+    env_ttl = os.environ.get("HERMES_OPENROUTER_CACHE_TTL", "").strip()
+    if env_ttl:
+        if env_ttl.isdigit():
+            ttl = int(env_ttl)
+            if 1 <= ttl <= 86400:
+                headers["X-OpenRouter-Cache-TTL"] = str(ttl)
+    else:
+        ttl = or_config.get("response_cache_ttl", 300)
+        if isinstance(ttl, (int, float)) and 1 <= ttl <= 86400:
+            headers["X-OpenRouter-Cache-TTL"] = str(int(ttl))
+
+    return headers
+
 # Vercel AI Gateway app attribution headers. HTTP-Referer maps to
 # referrerUrl and X-Title maps to appName in the gateway's analytics.
 from hermes_cli import __version__ as _HERMES_VERSION
@@ -512,7 +569,12 @@ class _CodexCompletionsAdapter:
                     # API allows it.
                     pass
                 else:
-                    effort = reasoning_cfg.get("effort", "medium")
+                    # Truthy-only check mirrors agent/transports/codex.py
+                    # build_kwargs(): falsy values (None, "", 0) fall back
+                    # to the default rather than being forwarded to the
+                    # Codex backend, which rejects e.g. {"effort": null}
+                    # with a 400.
+                    effort = reasoning_cfg.get("effort") or "medium"
                     # Codex backend rejects "minimal"; clamp to "low" to
                     # match the main-agent Codex transport behavior.
                     if effort == "minimal":
@@ -1149,23 +1211,23 @@ def _resolve_api_key_provider() -> Tuple[Optional[OpenAI], Optional[str]]:
 
 
 
-def _try_openrouter() -> Tuple[Optional[OpenAI], Optional[str]]:
+def _try_openrouter(explicit_api_key: str = None) -> Tuple[Optional[OpenAI], Optional[str]]:
     pool_present, entry = _select_pool_entry("openrouter")
     if pool_present:
-        or_key = _pool_runtime_api_key(entry)
+        or_key = explicit_api_key or _pool_runtime_api_key(entry)
         if not or_key:
             return None, None
         base_url = _pool_runtime_base_url(entry, OPENROUTER_BASE_URL) or OPENROUTER_BASE_URL
         logger.debug("Auxiliary client: OpenRouter via pool")
         return OpenAI(api_key=or_key, base_url=base_url,
-                       default_headers=_OR_HEADERS), _OPENROUTER_MODEL
+                       default_headers=build_or_headers()), _OPENROUTER_MODEL
 
-    or_key = os.getenv("OPENROUTER_API_KEY")
+    or_key = explicit_api_key or os.getenv("OPENROUTER_API_KEY")
     if not or_key:
         return None, None
     logger.debug("Auxiliary client: OpenRouter")
     return OpenAI(api_key=or_key, base_url=OPENROUTER_BASE_URL,
-                   default_headers=_OR_HEADERS), _OPENROUTER_MODEL
+                   default_headers=build_or_headers()), _OPENROUTER_MODEL
 
 
 def _describe_openrouter_unavailable() -> str:
@@ -1474,7 +1536,7 @@ def _build_codex_client(model: str) -> Tuple[Optional[Any], Optional[str]]:
     return CodexAuxiliaryClient(real_client, model), model
 
 
-def _try_anthropic() -> Tuple[Optional[Any], Optional[str]]:
+def _try_anthropic(explicit_api_key: str = None) -> Tuple[Optional[Any], Optional[str]]:
     try:
         from agent.anthropic_adapter import build_anthropic_client, resolve_anthropic_token
     except ImportError:
@@ -1484,10 +1546,10 @@ def _try_anthropic() -> Tuple[Optional[Any], Optional[str]]:
     if pool_present:
         if entry is None:
             return None, None
-        token = _pool_runtime_api_key(entry)
+        token = explicit_api_key or _pool_runtime_api_key(entry)
     else:
         entry = None
-        token = resolve_anthropic_token()
+        token = explicit_api_key or resolve_anthropic_token()
     if not token:
         return None, None
 
@@ -1588,6 +1650,39 @@ def _is_payment_error(exc: Exception) -> bool:
     return False
 
 
+def _is_rate_limit_error(exc: Exception) -> bool:
+    """Detect rate-limit errors that warrant provider fallback.
+
+    Returns True for HTTP 429 errors whose message indicates rate limiting
+    (as opposed to billing/quota exhaustion, which _is_payment_error handles).
+    Also catches OpenAI SDK RateLimitError instances that may not set
+    .status_code on the exception object.
+    """
+    status = getattr(exc, "status_code", None)
+    err_lower = str(exc).lower()
+
+    # OpenAI SDK's RateLimitError sometimes omits .status_code —
+    # detect by class name so we don't miss these.  (PR #8023 pattern)
+    if type(exc).__name__ == "RateLimitError":
+        return True
+
+    if status == 429:
+        # Distinguish rate-limit from billing: billing keywords are handled
+        # by _is_payment_error, everything else on 429 is a rate limit.
+        if any(kw in err_lower for kw in (
+            "rate limit", "rate_limit", "too many requests",
+            "try again", "retry after", "resets in",
+        )):
+            return True
+        # Generic 429 without billing keywords = likely a rate limit
+        if not any(kw in err_lower for kw in (
+            "credits", "insufficient funds", "billing",
+            "payment required", "can only afford",
+        )):
+            return True
+    return False
+
+
 def _is_connection_error(exc: Exception) -> bool:
     """Detect connection/network errors that warrant provider fallback.
 
@@ -1911,7 +2006,7 @@ def _to_async_client(sync_client, model: str, is_vision: bool = False):
     }
     sync_base_url = str(sync_client.base_url)
     if base_url_host_matches(sync_base_url, "openrouter.ai"):
-        async_kwargs["default_headers"] = dict(_OR_HEADERS)
+        async_kwargs["default_headers"] = build_or_headers()
     elif base_url_host_matches(sync_base_url, "api.githubcopilot.com"):
         from hermes_cli.copilot_auth import copilot_request_headers
 
@@ -1977,6 +2072,12 @@ def resolve_provider_client(
         (client, resolved_model) or (None, None) if auth is unavailable.
     """
     _validate_proxy_env_urls()
+    # Preserve the original provider name before alias normalization so a
+    # user-declared ``custom_providers`` entry whose name coincidentally
+    # matches a built-in alias (e.g. user names their custom provider "kimi"
+    # which aliases to "kimi-coding") is still reachable via the named-custom
+    # branch below.
+    original_provider = (provider or "").strip().lower()
     # Normalise aliases
     provider = _normalize_aux_provider(provider)
 
@@ -2047,9 +2148,9 @@ def resolve_provider_client(
         return (_to_async_client(client, final_model, is_vision=is_vision) if async_mode
                 else (client, final_model))
 
-    # ── OpenRouter ───────────────────────────────────────────────────
+    # ── OpenRouter ───────────────────────────────────────────
     if provider == "openrouter":
-        client, default = _try_openrouter()
+        client, default = _try_openrouter(explicit_api_key=explicit_api_key)
         if client is None:
             logger.warning(
                 "resolve_provider_client: openrouter requested but %s",
@@ -2163,7 +2264,18 @@ def resolve_provider_client(
     # ── Named custom providers (config.yaml providers dict / custom_providers list) ───
     try:
         from hermes_cli.runtime_provider import _get_named_custom_provider
-        custom_entry = _get_named_custom_provider(provider)
+        # When the raw requested name is an alias (``kimi`` → ``kimi-coding``)
+        # and the user defined a ``custom_providers`` entry under that alias
+        # name, the custom entry is the intended target — the built-in alias
+        # rewriting would otherwise hijack the request.  Only preferred when
+        # the raw name is an alias (not a canonical provider name) so custom
+        # entries that coincidentally match a canonical provider (e.g. ``nous``)
+        # still defer to the built-in per `_get_named_custom_provider`'s guard.
+        custom_entry = None
+        if original_provider and original_provider != provider:
+            custom_entry = _get_named_custom_provider(original_provider)
+        if custom_entry is None:
+            custom_entry = _get_named_custom_provider(provider)
         if custom_entry:
             custom_base = custom_entry.get("base_url", "").strip()
             custom_key = custom_entry.get("api_key", "").strip()
@@ -2264,7 +2376,7 @@ def resolve_provider_client(
 
     if pconfig.auth_type == "api_key":
         if provider == "anthropic":
-            client, default_model = _try_anthropic()
+            client, default_model = _try_anthropic(explicit_api_key=explicit_api_key)
             if client is None:
                 logger.warning("resolve_provider_client: anthropic requested but no Anthropic credentials found")
                 return None, None
@@ -2273,6 +2385,12 @@ def resolve_provider_client(
 
         creds = resolve_api_key_provider_credentials(provider)
         api_key = str(creds.get("api_key", "")).strip()
+        # Honour an explicit api_key override (e.g. from a fallback_model entry
+        # or a custom_providers entry) so callers that pass an explicit
+        # credential can authenticate against endpoints where no built-in
+        # credential is registered for this provider alias.
+        if explicit_api_key:
+            api_key = explicit_api_key.strip() or api_key
         if not api_key:
             tried_sources = list(pconfig.api_key_env_vars)
             if provider == "copilot":
@@ -2284,6 +2402,11 @@ def resolve_provider_client(
 
         raw_base_url = str(creds.get("base_url", "")).strip().rstrip("/") or pconfig.inference_base_url
         base_url = _to_openai_base_url(raw_base_url)
+        # Honour an explicit base_url override from the caller — used when a
+        # fallback_model entry (or custom_providers lookup) routes through a
+        # built-in provider name but targets a user-specified endpoint.
+        if explicit_base_url:
+            base_url = _to_openai_base_url(explicit_base_url.strip().rstrip("/"))
 
         default_model = _API_KEY_PROVIDER_AUX_MODELS.get(provider, "")
         final_model = _normalize_resolved_model(model or default_model, provider)
@@ -2565,8 +2688,11 @@ def resolve_vision_provider_client(
         return resolved_provider, sync_client, final_model
 
     if resolved_base_url:
+        provider_for_base_override = (
+            requested if requested and requested not in ("", "auto") else "custom"
+        )
         client, final_model = resolve_provider_client(
-            "custom",
+            provider_for_base_override,
             model=resolved_model,
             async_mode=async_mode,
             explicit_base_url=resolved_base_url,
@@ -2574,8 +2700,8 @@ def resolve_vision_provider_client(
             api_mode=resolved_api_mode,
         )
         if client is None:
-            return "custom", None, None
-        return "custom", client, final_model
+            return provider_for_base_override, None, None
+        return provider_for_base_override, client, final_model
 
     if requested == "auto":
         # Vision auto-detection order:
@@ -3041,8 +3167,14 @@ def _resolve_task_provider_model(
 
     if task:
         # Config.yaml is the primary source for per-task overrides.
-        if cfg_base_url:
+        if cfg_base_url and cfg_api_key:
+            # Both base_url and api_key explicitly set → custom endpoint.
             return "custom", resolved_model, cfg_base_url, cfg_api_key, resolved_api_mode
+        if cfg_base_url and cfg_provider and cfg_provider != "auto":
+            # base_url set without api_key but with a known provider — use
+            # the provider so it can resolve credentials from env vars
+            # (e.g. OPENROUTER_API_KEY) instead of locking into "custom".
+            return cfg_provider, resolved_model, cfg_base_url, None, resolved_api_mode
         if cfg_provider and cfg_provider != "auto":
             return cfg_provider, resolved_model, None, None, resolved_api_mode
 
@@ -3209,7 +3341,26 @@ def _build_call_kwargs(
             kwargs["max_tokens"] = max_tokens
 
     if tools:
-        kwargs["tools"] = tools
+        # Defensive dedup: providers like Google Vertex, Azure, and Bedrock
+        # reject requests with duplicate tool names (HTTP 400).  The upstream
+        # injection paths (run_agent.py) already dedup, but this guard
+        # converts a hard API failure into a warning if an upstream regression
+        # reintroduces duplicates.  See: #18478
+        _seen: set = set()
+        _deduped: list = []
+        for _t in tools:
+            _tname = (_t.get("function") or {}).get("name", "")
+            if _tname and _tname in _seen:
+                logger.warning(
+                    "_build_call_kwargs: duplicate tool name '%s' removed "
+                    "(provider=%s model=%s)",
+                    _tname, provider, model,
+                )
+                continue
+            if _tname:
+                _seen.add(_tname)
+            _deduped.append(_t)
+        kwargs["tools"] = _deduped
 
     # Provider-specific extra_body
     merged_extra = dict(extra_body or {})
@@ -3424,7 +3575,7 @@ def call_llm(
             except Exception as retry_err:
                 # If the max_tokens retry also hits a payment or connection
                 # error, fall through to the fallback chain below.
-                if not (_is_payment_error(retry_err) or _is_connection_error(retry_err)):
+                if not (_is_payment_error(retry_err) or _is_connection_error(retry_err) or _is_rate_limit_error(retry_err)):
                     raise
                 first_err = retry_err
 
@@ -3507,13 +3658,27 @@ def call_llm(
         # Codex/OAuth tokens that authenticate but whose endpoint is down,
         # and providers the user never configured that got picked up by
         # the auto-detection chain.
-        should_fallback = _is_payment_error(first_err) or _is_connection_error(first_err)
+        #
+        # ── Rate-limit fallback (#13579) ─────────────────────────────
+        # When the provider returns a 429 rate-limit (not billing), fall
+        # back to an alternative provider instead of exhausting retries
+        # against the same rate-limited endpoint.
+        should_fallback = (
+            _is_payment_error(first_err)
+            or _is_connection_error(first_err)
+            or _is_rate_limit_error(first_err)
+        )
         # Only try alternative providers when the user didn't explicitly
         # configure this task's provider.  Explicit provider = hard constraint;
         # auto (the default) = best-effort fallback chain.  (#7559)
         is_auto = resolved_provider in ("auto", "", None)
         if should_fallback and is_auto:
-            reason = "payment error" if _is_payment_error(first_err) else "connection error"
+            if _is_payment_error(first_err):
+                reason = "payment error"
+            elif _is_rate_limit_error(first_err):
+                reason = "rate limit"
+            else:
+                reason = "connection error"
             logger.info("Auxiliary %s: %s on %s (%s), trying fallback",
                         task or "call", reason, resolved_provider, first_err)
             fb_client, fb_model, fb_label = _try_payment_fallback(
@@ -3716,7 +3881,7 @@ async def async_call_llm(
             except Exception as retry_err:
                 # If the max_tokens retry also hits a payment or connection
                 # error, fall through to the fallback chain below.
-                if not (_is_payment_error(retry_err) or _is_connection_error(retry_err)):
+                if not (_is_payment_error(retry_err) or _is_connection_error(retry_err) or _is_rate_limit_error(retry_err)):
                     raise
                 first_err = retry_err
 
@@ -3785,11 +3950,20 @@ async def async_call_llm(
                     return _validate_llm_response(
                         await retry_client.chat.completions.create(**retry_kwargs), task)
 
-        # ── Payment / connection fallback (mirrors sync call_llm) ─────
-        should_fallback = _is_payment_error(first_err) or _is_connection_error(first_err)
+        # ── Payment / connection / rate-limit fallback (mirrors sync call_llm) ──
+        should_fallback = (
+            _is_payment_error(first_err)
+            or _is_connection_error(first_err)
+            or _is_rate_limit_error(first_err)
+        )
         is_auto = resolved_provider in ("auto", "", None)
         if should_fallback and is_auto:
-            reason = "payment error" if _is_payment_error(first_err) else "connection error"
+            if _is_payment_error(first_err):
+                reason = "payment error"
+            elif _is_rate_limit_error(first_err):
+                reason = "rate limit"
+            else:
+                reason = "connection error"
             logger.info("Auxiliary %s (async): %s on %s (%s), trying fallback",
                         task or "call", reason, resolved_provider, first_err)
             fb_client, fb_model, fb_label = _try_payment_fallback(

agent/context_compressor.py

@@ -344,6 +344,7 @@ class ContextCompressor(ContextEngine):
         self._last_aux_model_failure_model = None
         self._last_compression_savings_pct = 100.0
         self._ineffective_compression_count = 0
+        self._summary_failure_cooldown_until = 0.0  # transient errors must not block a fresh session
 
     def update_model(
         self,
@@ -538,7 +539,7 @@ class ContextCompressor(ContextEngine):
             # Token-budget approach: walk backward accumulating tokens
             accumulated = 0
             boundary = len(result)
-            min_protect = min(protect_tail_count, len(result) - 1)
+            min_protect = min(protect_tail_count, len(result))
             for i in range(len(result) - 1, -1, -1):
                 msg = result[i]
                 raw_content = msg.get("content") or ""
@@ -553,7 +554,16 @@ class ContextCompressor(ContextEngine):
                     break
                 accumulated += msg_tokens
                 boundary = i
-            prune_boundary = max(boundary, len(result) - min_protect)
+            # Translate the budget walk into a "protected count", apply the
+            # floor in count-space (where `max` reads naturally: protect at
+            # least `min_protect` messages or whatever the budget reserved,
+            # whichever is more), then convert back to a prune boundary.
+            # Doing this in index-space with `max` would invert the direction
+            # (smaller index = MORE protected), so a generous budget would
+            # silently get truncated back down to `min_protect`.
+            budget_protect_count = len(result) - boundary
+            protected_count = max(budget_protect_count, min_protect)
+            prune_boundary = len(result) - protected_count
         else:
             prune_boundary = len(result) - protect_tail_count
 
@@ -569,6 +579,8 @@ class ContextCompressor(ContextEngine):
             # Skip multimodal content (list of content blocks)
             if isinstance(content, list):
                 continue
+            if not isinstance(content, str):
+                continue
             if len(content) < 200:
                 continue
             h = hashlib.md5(content.encode("utf-8", errors="replace")).hexdigest()[:12]
@@ -588,6 +600,8 @@ class ContextCompressor(ContextEngine):
             # Skip multimodal content (list of content blocks)
             if isinstance(content, list):
                 continue
+            if not isinstance(content, str):
+                continue
             if not content or content == _PRUNED_TOOL_PLACEHOLDER:
                 continue
             # Skip already-deduplicated or previously-summarized results
@@ -903,15 +917,19 @@ The user has requested that this compaction PRIORITISE preserving all informatio
                 or "does not exist" in _err_str
                 or "no available channel" in _err_str
             )
+            _is_timeout = (
+                _status in (408, 429, 502, 504)
+                or "timeout" in _err_str
+            )
             if (
-                _is_model_not_found
+                (_is_model_not_found or _is_timeout)
                 and self.summary_model
                 and self.summary_model != self.model
                 and not getattr(self, "_summary_model_fallen_back", False)
             ):
                 self._summary_model_fallen_back = True
                 logging.warning(
-                    "Summary model '%s' not available (%s). "
+                    "Summary model '%s' unavailable (%s). "
                     "Falling back to main model '%s' for compression.",
                     self.summary_model, e, self.model,
                 )
@@ -975,15 +993,39 @@ The user has requested that this compaction PRIORITISE preserving all informatio
             return None
 
     @staticmethod
-    def _with_summary_prefix(summary: str) -> str:
-        """Normalize summary text to the current compaction handoff format."""
+    def _strip_summary_prefix(summary: str) -> str:
+        """Return summary body without the current or legacy handoff prefix."""
         text = (summary or "").strip()
-        for prefix in (LEGACY_SUMMARY_PREFIX, SUMMARY_PREFIX):
+        for prefix in (SUMMARY_PREFIX, LEGACY_SUMMARY_PREFIX):
             if text.startswith(prefix):
-                text = text[len(prefix):].lstrip()
-                break
+                return text[len(prefix):].lstrip()
+        return text
+
+    @classmethod
+    def _with_summary_prefix(cls, summary: str) -> str:
+        """Normalize summary text to the current compaction handoff format."""
+        text = cls._strip_summary_prefix(summary)
         return f"{SUMMARY_PREFIX}\n{text}" if text else SUMMARY_PREFIX
 
+    @staticmethod
+    def _is_context_summary_content(content: Any) -> bool:
+        text = _content_text_for_contains(content).lstrip()
+        return text.startswith(SUMMARY_PREFIX) or text.startswith(LEGACY_SUMMARY_PREFIX)
+
+    @classmethod
+    def _find_latest_context_summary(
+        cls,
+        messages: List[Dict[str, Any]],
+        start: int,
+        end: int,
+    ) -> tuple[Optional[int], str]:
+        """Find the newest handoff summary inside a compression window."""
+        for idx in range(end - 1, start - 1, -1):
+            content = messages[idx].get("content")
+            if cls._is_context_summary_content(content):
+                return idx, cls._strip_summary_prefix(_content_text_for_contains(content))
+        return None, ""
+
     # ------------------------------------------------------------------
     # Tool-call / tool-result pair integrity helpers
     # ------------------------------------------------------------------
@@ -992,8 +1034,8 @@ The user has requested that this compaction PRIORITISE preserving all informatio
     def _get_tool_call_id(tc) -> str:
         """Extract the call ID from a tool_call entry (dict or SimpleNamespace)."""
         if isinstance(tc, dict):
-            return tc.get("id", "")
-        return getattr(tc, "id", "") or ""
+            return tc.get("call_id", "") or tc.get("id", "") or ""
+        return getattr(tc, "call_id", "") or getattr(tc, "id", "") or ""
 
     def _sanitize_tool_pairs(self, messages: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
         """Fix orphaned tool_call / tool_result pairs after compression.
@@ -1290,6 +1332,15 @@ The user has requested that this compaction PRIORITISE preserving all informatio
             return messages
 
         turns_to_summarize = messages[compress_start:compress_end]
+        summary_idx, summary_body = self._find_latest_context_summary(
+            messages,
+            compress_start,
+            compress_end,
+        )
+        if summary_idx is not None:
+            if summary_body and not self._previous_summary:
+                self._previous_summary = summary_body
+            turns_to_summarize = messages[summary_idx + 1:compress_end]
 
         if not self.quiet_mode:
             logger.info(
@@ -1367,6 +1418,19 @@ The user has requested that this compaction PRIORITISE preserving all informatio
                 # Merge the summary into the first tail message instead
                 # of inserting a standalone message that breaks alternation.
                 _merge_summary_into_tail = True
+
+        # When the summary lands as a standalone role="user" message,
+        # weak models read the verbatim "## Active Task" quote of a past
+        # user request as fresh input (#11475, #14521). Append the explicit
+        # end marker — the same one used in the merge-into-tail path — so
+        # the model has a clear "summary above, not new input" signal.
+        if not _merge_summary_into_tail and summary_role == "user":
+            summary = (
+                summary
+                + "\n\n--- END OF CONTEXT SUMMARY — "
+                "respond to the message below, not the summary above ---"
+            )
+
         if not _merge_summary_into_tail:
             compressed.append({"role": summary_role, "content": summary})
 

agent/credential_pool.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import logging
+import os
 import random
 import threading
 import time
@@ -13,7 +14,7 @@ from datetime import datetime
 from typing import Any, Dict, List, Optional, Set, Tuple
 
 from hermes_constants import OPENROUTER_BASE_URL
-from hermes_cli.config import get_env_value
+from hermes_cli.config import get_env_value, load_env
 import hermes_cli.auth as auth_mod
 from hermes_cli.auth import (
     CODEX_ACCESS_TOKEN_REFRESH_SKEW_SECONDS,
@@ -1380,6 +1381,16 @@ def _seed_from_singletons(provider: str, entries: List[PooledCredential]) -> Tup
 def _seed_from_env(provider: str, entries: List[PooledCredential]) -> Tuple[bool, Set[str]]:
     changed = False
     active_sources: Set[str] = set()
+
+    # Prefer ~/.hermes/.env over os.environ — the user's config file is the
+    # authoritative source for Hermes credentials. Stale env vars from parent
+    # processes (Codex CLI, test scripts, etc.) should not override deliberate
+    # changes to the .env file.
+    def _get_env_prefer_dotenv(key: str) -> str:
+        env_file = load_env()
+        val = env_file.get(key) or os.environ.get(key) or ""
+        return val.strip()
+
     # Honour user suppression — `hermes auth remove <provider> <N>` for an
     # env-seeded credential marks the env:<VAR> source as suppressed so it
     # won't be re-seeded from the user's shell environment or ~/.hermes/.env.
@@ -1391,8 +1402,8 @@ def _seed_from_env(provider: str, entries: List[PooledCredential]) -> Tuple[bool
         def _is_source_suppressed(_p, _s):  # type: ignore[misc]
             return False
     if provider == "openrouter":
-        # Check both os.environ and ~/.hermes/.env file
-        token = (get_env_value("OPENROUTER_API_KEY") or "").strip()
+        # Prefer ~/.hermes/.env over os.environ
+        token = _get_env_prefer_dotenv("OPENROUTER_API_KEY")
         if token:
             source = "env:OPENROUTER_API_KEY"
             if _is_source_suppressed(provider, source):
@@ -1418,7 +1429,7 @@ def _seed_from_env(provider: str, entries: List[PooledCredential]) -> Tuple[bool
 
     env_url = ""
     if pconfig.base_url_env_var:
-        env_url = (get_env_value(pconfig.base_url_env_var) or "").strip().rstrip("/")
+        env_url = _get_env_prefer_dotenv(pconfig.base_url_env_var).rstrip("/")
 
     env_vars = list(pconfig.api_key_env_vars)
     if provider == "anthropic":
@@ -1429,8 +1440,8 @@ def _seed_from_env(provider: str, entries: List[PooledCredential]) -> Tuple[bool
         ]
 
     for env_var in env_vars:
-        # Check both os.environ and ~/.hermes/.env file
-        token = (get_env_value(env_var) or "").strip()
+        # Prefer ~/.hermes/.env over os.environ
+        token = _get_env_prefer_dotenv(env_var)
         if not token:
             continue
         source = f"env:{env_var}"

agent/curator.py

@@ -24,11 +24,12 @@ from __future__ import annotations
 import json
 import logging
 import os
+import re
 import tempfile
 import threading
 from datetime import datetime, timedelta, timezone
 from pathlib import Path
-from typing import Any, Callable, Dict, List, Optional, Set
+from typing import Any, Callable, Dict, List, NamedTuple, Optional, Set
 
 from hermes_constants import get_hermes_home
 from tools import skill_usage
@@ -36,6 +37,22 @@ from tools import skill_usage
 logger = logging.getLogger(__name__)
 
 
+def _strip_aux_credential(value: Any) -> Optional[str]:
+    if value is None:
+        return None
+    text = str(value).strip()
+    return text or None
+
+
+class _ReviewRuntimeBinding(NamedTuple):
+    """Provider/model for the curator review fork plus optional per-slot overrides."""
+
+    provider: str
+    model: str
+    explicit_api_key: Optional[str]
+    explicit_base_url: Optional[str]
+
+
 DEFAULT_INTERVAL_HOURS = 24 * 7  # 7 days
 DEFAULT_MIN_IDLE_HOURS = 2
 DEFAULT_STALE_AFTER_DAYS = 30
@@ -55,6 +72,7 @@ def _default_state() -> Dict[str, Any]:
         "last_run_at": None,
         "last_run_duration_seconds": None,
         "last_run_summary": None,
+        "last_report_path": None,
         "paused": False,
         "run_count": 0,
     }
@@ -183,7 +201,16 @@ def should_run_now(now: Optional[datetime] = None) -> bool:
     Gates:
       - curator.enabled == True
       - not paused
-      - last_run_at missing, OR older than interval_hours
+      - last_run_at present AND older than interval_hours
+
+    First-run behavior: when there is no ``last_run_at`` (fresh install, or
+    install that predates the curator), we DO NOT run immediately. The
+    curator is designed to run after at least ``interval_hours`` (7 days by
+    default) of skill activity, not on the first background tick after
+    ``hermes update``. On first observation we seed ``last_run_at`` to "now"
+    and defer the first real pass by one full interval. Users who want to
+    run it sooner can always invoke ``hermes curator run`` (with or without
+    ``--dry-run``) explicitly — that path bypasses this gate.
 
     The idle check (min_idle_hours) is applied at the call site where we know
     whether an agent is actively running — here we only enforce the static
@@ -197,7 +224,21 @@ def should_run_now(now: Optional[datetime] = None) -> bool:
     state = load_state()
     last = _parse_iso(state.get("last_run_at"))
     if last is None:
-        return True
+        # Never run before. Seed state so we wait a full interval before the
+        # first real pass. Report-only; do not auto-mutate the library the
+        # very first time a gateway ticks after an update.
+        if now is None:
+            now = datetime.now(timezone.utc)
+        try:
+            state["last_run_at"] = now.isoformat()
+            state["last_run_summary"] = (
+                "deferred first run — curator seeded, will run after one "
+                "interval; use `hermes curator run --dry-run` to preview now"
+            )
+            save_state(state)
+        except Exception as e:  # pragma: no cover — best-effort persistence
+            logger.debug("Failed to seed curator last_run_at: %s", e)
+        return False
 
     if now is None:
         now = datetime.now(timezone.utc)
@@ -258,6 +299,33 @@ def apply_automatic_transitions(now: Optional[datetime] = None) -> Dict[str, int
 # Review prompt for the forked agent
 # ---------------------------------------------------------------------------
 
+CURATOR_DRY_RUN_BANNER = (
+    "═══════════════════════════════════════════════════════════════\n"
+    "DRY-RUN — REPORT ONLY. DO NOT MUTATE THE SKILL LIBRARY.\n"
+    "═══════════════════════════════════════════════════════════════\n"
+    "\n"
+    "This is a PREVIEW pass. Follow every instruction below EXCEPT:\n"
+    "\n"
+    "  • DO NOT call skill_manage with action=patch, create, delete, "
+    "write_file, or remove_file.\n"
+    "  • DO NOT call terminal to mv skill directories into .archive/.\n"
+    "  • DO NOT call terminal to mv, cp, rm, or rewrite any file under "
+    "~/.hermes/skills/.\n"
+    "  • skills_list and skill_view are FINE — read as much as you need.\n"
+    "\n"
+    "Your output IS the deliverable. Produce the exact same "
+    "human-readable summary and structured YAML block you would "
+    "produce on a live run — but describe the actions you WOULD take, "
+    "not actions you took. A downstream reviewer will read the report "
+    "and decide whether to approve a live run with "
+    "`hermes curator run` (no flag).\n"
+    "\n"
+    "If you accidentally take a mutating action, say so explicitly in "
+    "the summary so the reviewer can revert it.\n"
+    "═══════════════════════════════════════════════════════════════"
+)
+
+
 CURATOR_REVIEW_PROMPT = (
     "You are running as Hermes' background skill CURATOR. This is an "
     "UMBRELLA-BUILDING consolidation pass, not a passive audit and not a "
@@ -336,6 +404,11 @@ CURATOR_REVIEW_PROMPT = (
     "  - skill_manage action=write_file — add a references/, templates/, "
     "or scripts/ file under an existing skill (the skill must already "
     "exist)\n"
+    "  - skill_manage action=delete     — archive a skill. MUST pass "
+    "`absorbed_into=<umbrella>` when you've merged its content into another "
+    "skill, or `absorbed_into=\"\"` when you're truly pruning with no "
+    "forwarding target. This drives cron-job skill-reference migration — "
+    "guessing from your YAML summary after the fact is fragile.\n"
     "  - terminal                       — mv a sibling into the archive "
     "OR move its content into a support subfile\n\n"
     "'keep' is a legitimate decision ONLY when the skill is already a "
@@ -397,6 +470,24 @@ def _reports_root() -> Path:
     return root
 
 
+def _needle_in_path_component(needle: str, path: str) -> bool:
+    """Check if *needle* is a complete filename stem or directory name in *path*.
+
+    Unlike simple substring matching, this avoids false positives where short
+    skill names are embedded in longer filenames (e.g. "api" matching
+    "references/api-design.md").  Hyphens and underscores are normalised so
+    "open-webui-setup" matches "open_webui_setup.md".
+    """
+    norm_needle = needle.replace("-", "_")
+    for part in path.replace("\\", "/").split("/"):
+        if not part:
+            continue
+        stem = part.rsplit(".", 1)[0] if "." in part else part
+        if stem.replace("-", "_") == norm_needle:
+            return True
+    return False
+
+
 def _classify_removed_skills(
     removed: List[str],
     added: List[str],
@@ -475,15 +566,29 @@ def _classify_removed_skills(
                 continue
 
             # Look for the removed skill's name in file_path / content / raw.
-            haystacks: List[str] = []
+            # Matching strategy differs by field type:
+            #   file_path — needle must be a complete path component
+            #     (filename stem or directory name), so "api" does NOT
+            #     falsely match "references/api-design.md".
+            #   content fields — word-boundary regex so "test" does NOT
+            #     falsely match "latest" or "testing".
+            haystacks: List[tuple[str, str]] = []
             for key in ("file_path", "file_content", "content", "new_string", "_raw"):
                 v = args.get(key)
                 if isinstance(v, str):
-                    haystacks.append(v)
+                    haystacks.append((key, v))
             hit = False
-            for hay in haystacks:
+            for key, hay in haystacks:
                 for needle in needles:
-                    if needle and needle in hay:
+                    if not needle:
+                        continue
+                    if key == "file_path":
+                        matched = _needle_in_path_component(needle, hay)
+                    else:
+                        matched = bool(
+                            re.search(rf'\b{re.escape(needle)}\b', hay)
+                        )
+                    if matched:
                         hit = True
                         evidence = (
                             f"skill_manage action={args.get('action', '?')} "
@@ -586,15 +691,76 @@ def _parse_structured_summary(
     return out
 
 
+def _extract_absorbed_into_declarations(
+    tool_calls: List[Dict[str, Any]],
+) -> Dict[str, Dict[str, Any]]:
+    """Walk this run's tool calls and extract model-declared absorption targets.
+
+    The curator prompt requires every ``skill_manage(action='delete')`` call
+    to pass ``absorbed_into=<umbrella>`` when consolidating, or
+    ``absorbed_into=""`` when truly pruning. This is the single authoritative
+    signal for classification — the model's own declaration at the moment of
+    deletion, which beats both post-hoc YAML summary parsing and substring
+    heuristics on other tool calls.
+
+    Returns ``{skill_name: {"into": "<umbrella>" | "", "declared": True}}``.
+    Entries with ``into == ""`` are explicit prunings.
+    Skills without a ``skill_manage(delete)`` call, or with one that omitted
+    ``absorbed_into``, are not in the returned dict — caller falls back to
+    the existing heuristic/YAML logic for those (backward compat with older
+    curator runs and any callers that don't populate the arg).
+    """
+    out: Dict[str, Dict[str, Any]] = {}
+    for tc in tool_calls or []:
+        if not isinstance(tc, dict):
+            continue
+        if tc.get("name") != "skill_manage":
+            continue
+        raw = tc.get("arguments") or ""
+        args: Dict[str, Any] = {}
+        if isinstance(raw, dict):
+            args = raw
+        elif isinstance(raw, str):
+            try:
+                args = json.loads(raw)
+            except Exception:
+                continue
+        if not isinstance(args, dict):
+            continue
+        if args.get("action") != "delete":
+            continue
+        name = args.get("name")
+        if not isinstance(name, str) or not name.strip():
+            continue
+        # absorbed_into must be present (even empty string is meaningful);
+        # missing key means the model didn't declare intent.
+        if "absorbed_into" not in args:
+            continue
+        target = args.get("absorbed_into")
+        if target is None:
+            continue
+        if not isinstance(target, str):
+            continue
+        out[name.strip()] = {"into": target.strip(), "declared": True}
+    return out
+
+
 def _reconcile_classification(
     removed: List[str],
     heuristic: Dict[str, List[Dict[str, Any]]],
     model_block: Dict[str, List[Dict[str, str]]],
     destinations: Set[str],
+    absorbed_declarations: Optional[Dict[str, Dict[str, Any]]] = None,
 ) -> Dict[str, List[Dict[str, Any]]]:
     """Merge heuristic (tool-call evidence) with the model's structured block.
 
-    Rules:
+    Rules (evaluated in order; first match wins):
+    - **Model-declared `absorbed_into` at delete time is authoritative.** Any
+      entry in ``absorbed_declarations`` beats every other signal. This is
+      the model telling us directly, at the moment of deletion, what it did.
+      ``into != ""`` and target exists → consolidated. ``into == ""`` →
+      pruned. ``into != ""`` but target doesn't exist → hallucination; fall
+      through to the usual signals.
     - Model-declared consolidation wins when its ``into`` target exists
       in ``destinations`` (survived or newly-created). This gives the
       model authority over intent + rationale.
@@ -615,6 +781,8 @@ def _reconcile_classification(
     model_cons = {e["from"]: e for e in model_block.get("consolidations", [])}
     model_pruned = {e["name"]: e for e in model_block.get("prunings", [])}
 
+    declared = absorbed_declarations or {}
+
     consolidated: List[Dict[str, Any]] = []
     pruned: List[Dict[str, Any]] = []
 
@@ -622,6 +790,36 @@ def _reconcile_classification(
         mc = model_cons.get(name)
         mp = model_pruned.get(name)
         hc = heur_cons.get(name)
+        dec = declared.get(name)
+
+        # Authoritative: model declared `absorbed_into` at the delete call.
+        if dec is not None:
+            into_claim = dec.get("into", "")
+            if into_claim and into_claim in destinations:
+                entry: Dict[str, Any] = {
+                    "name": name,
+                    "into": into_claim,
+                    "source": "absorbed_into (model-declared at delete)",
+                    "reason": (mc.get("reason") or "") if mc else "",
+                }
+                if hc and hc.get("evidence"):
+                    entry["evidence"] = hc["evidence"]
+                consolidated.append(entry)
+                continue
+            if into_claim == "":
+                # Explicit prune declaration
+                pruned.append({
+                    "name": name,
+                    "source": "absorbed_into=\"\" (model-declared prune)",
+                    "reason": (mp.get("reason") or "") if mp else "",
+                })
+                continue
+            # into_claim is non-empty but target doesn't exist: the model
+            # named a nonexistent umbrella at delete time. The tool already
+            # rejects this at the skill_manage layer, so we shouldn't see it
+            # in practice — but if it slips through (e.g. the umbrella was
+            # deleted LATER in the same run), fall through to the usual
+            # signals rather than trusting a broken reference.
 
         # Model says consolidated — trust it if the destination is real.
         if mc and mc.get("into") in destinations:
@@ -757,15 +955,57 @@ def _write_run_report(
     )
     model_block = _parse_structured_summary(llm_meta.get("final", "") or "")
     destinations = set(after_names) | set(added or [])
+    # Authoritative signal: extract per-delete `absorbed_into` declarations
+    # from this run's tool calls. These beat both the YAML summary block and
+    # the substring heuristic — the model is telling us directly, at the
+    # moment of deletion, whether each archived skill was consolidated
+    # (into=<umbrella>) or pruned (into="").
+    absorbed_declarations = _extract_absorbed_into_declarations(
+        llm_meta.get("tool_calls", []) or []
+    )
     classification = _reconcile_classification(
         removed=removed,
         heuristic=heuristic,
         model_block=model_block,
         destinations=destinations,
+        absorbed_declarations=absorbed_declarations,
     )
     consolidated = classification["consolidated"]
     pruned = classification["pruned"]
 
+    # Rewrite cron job skill references. When the curator consolidates
+    # skill X into umbrella Y, any cron job that lists X fails to load
+    # it at run time — the scheduler skips it and the job runs without
+    # the instructions it was scheduled to follow. Rewriting the
+    # references in-place keeps scheduled jobs working across
+    # consolidation passes. Best-effort: never let a cron-module issue
+    # break the curator.
+    cron_rewrites: Dict[str, Any] = {"rewrites": [], "jobs_updated": 0, "jobs_scanned": 0}
+    try:
+        consolidated_map = {
+            e["name"]: e["into"]
+            for e in consolidated
+            if isinstance(e, dict) and e.get("name") and e.get("into")
+        }
+        pruned_names = [
+            e["name"] for e in pruned
+            if isinstance(e, dict) and e.get("name")
+        ]
+        if consolidated_map or pruned_names:
+            from cron.jobs import rewrite_skill_refs as _rewrite_cron_refs
+            cron_rewrites = _rewrite_cron_refs(
+                consolidated=consolidated_map,
+                pruned=pruned_names,
+            )
+    except Exception as e:
+        logger.debug("Curator cron skill rewrite failed: %s", e, exc_info=True)
+        cron_rewrites = {
+            "rewrites": [],
+            "jobs_updated": 0,
+            "jobs_scanned": 0,
+            "error": str(e),
+        }
+
     payload = {
         "started_at": started_at.isoformat(),
         "duration_seconds": round(elapsed_seconds, 2),
@@ -781,6 +1021,7 @@ def _write_run_report(
             "consolidated_this_run": len(consolidated),
             "pruned_this_run": len(pruned),
             "state_transitions": len(transitions),
+            "cron_jobs_rewritten": int(cron_rewrites.get("jobs_updated", 0)),
             "tool_calls_total": sum(tc_counts.values()),
         },
         "tool_call_counts": tc_counts,
@@ -790,6 +1031,7 @@ def _write_run_report(
         "pruned_names": [p["name"] for p in pruned],
         "added": added,
         "state_transitions": transitions,
+        "cron_rewrites": cron_rewrites,
         "llm_final": llm_meta.get("final", ""),
         "llm_summary": llm_meta.get("summary", ""),
         "llm_error": llm_meta.get("error"),
@@ -812,6 +1054,17 @@ def _write_run_report(
     except Exception as e:
         logger.debug("Curator REPORT.md write failed: %s", e)
 
+    # cron_rewrites.json — only when at least one job was touched, to
+    # keep run dirs uncluttered for the common no-op case.
+    try:
+        if int(cron_rewrites.get("jobs_updated", 0)) > 0:
+            (run_dir / "cron_rewrites.json").write_text(
+                json.dumps(cron_rewrites, indent=2, ensure_ascii=False) + "\n",
+                encoding="utf-8",
+            )
+    except Exception as e:
+        logger.debug("Curator cron_rewrites.json write failed: %s", e)
+
     return run_dir
 
 
@@ -942,6 +1195,39 @@ def _render_report_markdown(p: Dict[str, Any]) -> str:
             lines.append(f"- `{t.get('name')}`: {t.get('from')} → {t.get('to')}")
         lines.append("")
 
+    # Cron job rewrites — show which scheduled jobs had their skill
+    # references updated so users can audit that the auto-rewrite did
+    # the right thing. Only present when at least one job changed.
+    cron_rw = p.get("cron_rewrites") or {}
+    cron_rewrites_list = cron_rw.get("rewrites") or []
+    if cron_rewrites_list:
+        lines.append(f"### Cron job skill references rewritten ({len(cron_rewrites_list)})\n")
+        lines.append(
+            "_Cron jobs that referenced a consolidated or pruned skill were "
+            "updated in-place so they keep loading the right instructions "
+            "on their next run. See `cron_rewrites.json` for the full record._\n"
+        )
+        SHOW = 25
+        for entry in cron_rewrites_list[:SHOW]:
+            job_name = entry.get("job_name") or entry.get("job_id") or "?"
+            before = entry.get("before") or []
+            after = entry.get("after") or []
+            mapped = entry.get("mapped") or {}
+            dropped = entry.get("dropped") or []
+            lines.append(
+                f"- `{job_name}`: `{', '.join(before)}` → `{', '.join(after) or '(none)'}`"
+            )
+            for old, new in mapped.items():
+                lines.append(f"    - `{old}` → `{new}` (consolidated)")
+            for name in dropped:
+                lines.append(f"    - `{name}` dropped (pruned)")
+        if len(cron_rewrites_list) > SHOW:
+            lines.append(
+                f"- … and {len(cron_rewrites_list) - SHOW} more "
+                "(see `cron_rewrites.json`)"
+            )
+        lines.append("")
+
     # Full LLM final response
     final = (p.get("llm_final") or "").strip()
     if final:
@@ -992,6 +1278,7 @@ def _render_candidate_list() -> str:
 def run_curator_review(
     on_summary: Optional[Callable[[str], None]] = None,
     synchronous: bool = False,
+    dry_run: bool = False,
 ) -> Dict[str, Any]:
     """Execute a single curator review pass.
 
@@ -1004,9 +1291,43 @@ def run_curator_review(
 
     If *synchronous* is True, the LLM review runs in the calling thread; the
     default is to spawn a daemon thread so the caller returns immediately.
+
+    If *dry_run* is True, the automatic stale/archive transitions are SKIPPED
+    and the LLM review pass is instructed to produce a report only — no
+    skill_manage mutations, no terminal archive moves. The REPORT.md still
+    gets written and ``state.last_report_path`` still records it so users
+    can read what the curator WOULD have done.
     """
     start = datetime.now(timezone.utc)
-    counts = apply_automatic_transitions(now=start)
+    if dry_run:
+        # Count candidates without mutating state.
+        try:
+            report = skill_usage.agent_created_report()
+            counts = {
+                "checked": len(report),
+                "marked_stale": 0,
+                "archived": 0,
+                "reactivated": 0,
+            }
+        except Exception:
+            counts = {"checked": 0, "marked_stale": 0, "archived": 0, "reactivated": 0}
+    else:
+        # Pre-mutation snapshot — best-effort, never blocks the run. A
+        # failed snapshot logs at debug and continues (the alternative is
+        # that a transient disk issue silently disables curator forever,
+        # which is worse). Users who want to require snapshots can disable
+        # curator entirely until they can fix disk space.
+        try:
+            from agent import curator_backup
+            snap = curator_backup.snapshot_skills(reason="pre-curator-run")
+            if snap is not None and on_summary:
+                try:
+                    on_summary(f"curator: snapshot created ({snap.name})")
+                except Exception:
+                    pass
+        except Exception as e:
+            logger.debug("Curator pre-run snapshot failed: %s", e, exc_info=True)
+        counts = apply_automatic_transitions(now=start)
 
     auto_summary_parts = []
     if counts["marked_stale"]:
@@ -1018,11 +1339,16 @@ def run_curator_review(
     auto_summary = ", ".join(auto_summary_parts) if auto_summary_parts else "no changes"
 
     # Persist state before the LLM pass so a crash mid-review still records
-    # the run and doesn't immediately re-trigger.
+    # the run and doesn't immediately re-trigger. In dry-run we do NOT bump
+    # last_run_at or run_count — a preview shouldn't push the next scheduled
+    # real pass out. We still record a summary so `hermes curator status`
+    # shows that a preview ran.
     state = load_state()
-    state["last_run_at"] = start.isoformat()
-    state["run_count"] = int(state.get("run_count", 0)) + 1
-    state["last_run_summary"] = f"auto: {auto_summary}"
+    if not dry_run:
+        state["last_run_at"] = start.isoformat()
+        state["run_count"] = int(state.get("run_count", 0)) + 1
+    prefix = "dry-run auto: " if dry_run else "auto: "
+    state["last_run_summary"] = f"{prefix}{auto_summary}"
     save_state(state)
 
     def _llm_pass():
@@ -1038,7 +1364,7 @@ def run_curator_review(
         try:
             candidate_list = _render_candidate_list()
             if "No agent-created skills" in candidate_list:
-                final_summary = f"auto: {auto_summary}; llm: skipped (no candidates)"
+                final_summary = f"{prefix}{auto_summary}; llm: skipped (no candidates)"
                 llm_meta = {
                     "final": "",
                     "summary": "skipped (no candidates)",
@@ -1048,14 +1374,21 @@ def run_curator_review(
                     "error": None,
                 }
             else:
-                prompt = f"{CURATOR_REVIEW_PROMPT}\n\n{candidate_list}"
+                if dry_run:
+                    prompt = (
+                        f"{CURATOR_DRY_RUN_BANNER}\n\n"
+                        f"{CURATOR_REVIEW_PROMPT}\n\n"
+                        f"{candidate_list}"
+                    )
+                else:
+                    prompt = f"{CURATOR_REVIEW_PROMPT}\n\n{candidate_list}"
                 llm_meta = _run_llm_review(prompt)
                 final_summary = (
-                    f"auto: {auto_summary}; llm: {llm_meta.get('summary', 'no change')}"
+                    f"{prefix}{auto_summary}; llm: {llm_meta.get('summary', 'no change')}"
                 )
         except Exception as e:
             logger.debug("Curator LLM pass failed: %s", e, exc_info=True)
-            final_summary = f"auto: {auto_summary}; llm: error ({e})"
+            final_summary = f"{prefix}{auto_summary}; llm: error ({e})"
             llm_meta = {
                 "final": "",
                 "summary": f"error ({e})",
@@ -1114,6 +1447,52 @@ def run_curator_review(
     }
 
 
+def _resolve_review_runtime(cfg: Dict[str, Any]) -> _ReviewRuntimeBinding:
+    """Resolve provider/model and per-slot credentials for the curator review fork.
+
+    Same precedence as `_resolve_review_model()`. Non-empty ``api_key`` /
+    ``base_url`` from the active slot are returned as explicit overrides so
+    ``resolve_runtime_provider`` does not silently reuse the main chat
+    credential chain for a routed auxiliary model.
+    """
+    _main = cfg.get("model", {}) if isinstance(cfg.get("model"), dict) else {}
+    _main_provider = _main.get("provider") or "auto"
+    _main_model = _main.get("default") or _main.get("model") or ""
+
+    # 1. Canonical aux task slot
+    _aux = cfg.get("auxiliary", {}) if isinstance(cfg.get("auxiliary"), dict) else {}
+    _cur_task = _aux.get("curator", {}) if isinstance(_aux.get("curator"), dict) else {}
+    _task_provider = (_cur_task.get("provider") or "").strip() or None
+    _task_model = (_cur_task.get("model") or "").strip() or None
+    if _task_provider and _task_provider != "auto" and _task_model:
+        return _ReviewRuntimeBinding(
+            _task_provider,
+            _task_model,
+            _strip_aux_credential(_cur_task.get("api_key")),
+            _strip_aux_credential(_cur_task.get("base_url")),
+        )
+
+    # 2. Legacy curator.auxiliary.{provider,model} (deprecated, pre-unification)
+    _cur = cfg.get("curator", {}) if isinstance(cfg.get("curator"), dict) else {}
+    _legacy = _cur.get("auxiliary", {}) if isinstance(_cur.get("auxiliary"), dict) else {}
+    _legacy_provider = _legacy.get("provider") or None
+    _legacy_model = _legacy.get("model") or None
+    if _legacy_provider and _legacy_model:
+        logger.info(
+            "curator: using deprecated curator.auxiliary.{provider,model} "
+            "config — please migrate to auxiliary.curator.{provider,model}"
+        )
+        return _ReviewRuntimeBinding(
+            str(_legacy_provider),
+            str(_legacy_model),
+            _strip_aux_credential(_legacy.get("api_key")),
+            _strip_aux_credential(_legacy.get("base_url")),
+        )
+
+    # 3. Fall through to the main chat model
+    return _ReviewRuntimeBinding(_main_provider, _main_model, None, None)
+
+
 def _resolve_review_model(cfg: Dict[str, Any]) -> tuple[str, str]:
     """Pick (provider, model) for the curator review fork.
 
@@ -1129,32 +1508,8 @@ def _resolve_review_model(cfg: Dict[str, Any]) -> tuple[str, str]:
       2. Legacy ``curator.auxiliary.{provider,model}`` when both are set
       3. Main ``model.{provider,default/model}`` pair
     """
-    _main = cfg.get("model", {}) if isinstance(cfg.get("model"), dict) else {}
-    _main_provider = _main.get("provider") or "auto"
-    _main_model = _main.get("default") or _main.get("model") or ""
-
-    # 1. Canonical aux task slot
-    _aux = cfg.get("auxiliary", {}) if isinstance(cfg.get("auxiliary"), dict) else {}
-    _cur_task = _aux.get("curator", {}) if isinstance(_aux.get("curator"), dict) else {}
-    _task_provider = (_cur_task.get("provider") or "").strip() or None
-    _task_model = (_cur_task.get("model") or "").strip() or None
-    if _task_provider and _task_provider != "auto" and _task_model:
-        return _task_provider, _task_model
-
-    # 2. Legacy curator.auxiliary.{provider,model} (deprecated, pre-unification)
-    _cur = cfg.get("curator", {}) if isinstance(cfg.get("curator"), dict) else {}
-    _legacy = _cur.get("auxiliary", {}) if isinstance(_cur.get("auxiliary"), dict) else {}
-    _legacy_provider = _legacy.get("provider") or None
-    _legacy_model = _legacy.get("model") or None
-    if _legacy_provider and _legacy_model:
-        logger.info(
-            "curator: using deprecated curator.auxiliary.{provider,model} "
-            "config — please migrate to auxiliary.curator.{provider,model}"
-        )
-        return _legacy_provider, _legacy_model
-
-    # 3. Fall through to the main chat model
-    return _main_provider, _main_model
+    b = _resolve_review_runtime(cfg)
+    return b.provider, b.model
 
 
 def _run_llm_review(prompt: str) -> Dict[str, Any]:
@@ -1193,10 +1548,10 @@ def _run_llm_review(prompt: str) -> Dict[str, Any]:
     # arguments hits an auto-resolution path that fails for OAuth-only
     # providers and for pool-backed credentials.
     #
-    # `_resolve_review_model()` honors `auxiliary.curator.{provider,model}`
+    # `_resolve_review_runtime()` honors `auxiliary.curator.{provider,model,...}`
     # (canonical aux-task slot, wired through `hermes model` → auxiliary
     # picker and the dashboard Models tab), with a legacy fallback to
-    # `curator.auxiliary.{provider,model}`. See docs/user-guide/features/curator.md.
+    # `curator.auxiliary.{provider,model,...}`. See docs/user-guide/features/curator.md.
     _api_key = None
     _base_url = None
     _api_mode = None
@@ -1206,9 +1561,13 @@ def _run_llm_review(prompt: str) -> Dict[str, Any]:
         from hermes_cli.config import load_config
         from hermes_cli.runtime_provider import resolve_runtime_provider
         _cfg = load_config()
-        _provider, _model_name = _resolve_review_model(_cfg)
+        _binding = _resolve_review_runtime(_cfg)
+        _provider, _model_name = _binding.provider, _binding.model
         _rp = resolve_runtime_provider(
-            requested=_provider, target_model=_model_name
+            requested=_provider,
+            target_model=_model_name,
+            explicit_api_key=_binding.explicit_api_key,
+            explicit_base_url=_binding.explicit_base_url,
         )
         _api_key = _rp.get("api_key")
         _base_url = _rp.get("base_url")

agent/curator_backup.py

@@ -0,0 +1,693 @@
+"""Curator snapshot + rollback.
+
+A pre-run snapshot of ``~/.hermes/skills/`` (excluding ``.curator_backups/``
+itself) is taken before any mutating curator pass. Snapshots are tar.gz
+files under ``~/.hermes/skills/.curator_backups/<utc-iso>/`` with a
+companion ``manifest.json`` describing the snapshot (reason, time, size,
+counted skill files). Rollback picks a snapshot, moves the current
+``skills/`` tree aside into another snapshot so even the rollback itself
+is undoable, then extracts the chosen snapshot into place.
+
+The snapshot does NOT include:
+  - ``.curator_backups/`` (would recurse)
+  - ``.hub/`` (hub-installed skills — managed by the hub, not us)
+
+It DOES include:
+  - all SKILL.md files + their directories (``scripts/``, ``references/``,
+    ``templates/``, ``assets/``)
+  - ``.usage.json`` (usage telemetry — needed to rehydrate state cleanly)
+  - ``.archive/`` (so rollback restores previously-archived skills too)
+  - ``.curator_state`` (so rolling back also restores the last-run-at
+    pointer — otherwise the curator would immediately re-fire on the next
+    tick)
+  - ``.bundled_manifest`` (so protection markers stay consistent)
+
+Alongside the skills tarball, each snapshot also captures a copy of
+``~/.hermes/cron/jobs.json`` as ``cron-jobs.json`` when it exists. Cron
+jobs reference skills by name in their ``skills``/``skill`` fields; the
+curator's consolidation pass rewrites those in place via
+``cron.jobs.rewrite_skill_refs()``. Without capturing the pre-run state,
+rolling back the skills tree would leave cron jobs pointing at the
+umbrella skills even though the narrow skills they were originally
+configured with have been restored. We store the whole jobs.json for
+fidelity but rollback only touches the ``skills``/``skill`` fields — the
+rest (schedule, next_run_at, enabled, prompt, etc.) is live state and
+we leave it alone.
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+import os
+import re
+import shutil
+import tarfile
+import tempfile
+import time
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any, Dict, List, Optional, Tuple
+
+from hermes_constants import get_hermes_home
+
+logger = logging.getLogger(__name__)
+
+
+DEFAULT_KEEP = 5
+
+# Entries under skills/ that should NEVER be rolled up into a snapshot.
+# .hub/ is managed by the skills hub; rolling it back would break lockfile
+# invariants. .curator_backups is the backup dir itself — recursion bomb.
+_EXCLUDE_TOP_LEVEL = {".curator_backups", ".hub"}
+
+# Snapshot id regex: UTC ISO with colons replaced by dashes so the filename
+# is portable (Windows-safe). An optional ``-NN`` suffix handles two
+# snapshots landing in the same wallclock second.
+_ID_RE = re.compile(r"^\d{4}-\d{2}-\d{2}T\d{2}-\d{2}-\d{2}Z(-\d{2})?$")
+
+
+def _backups_dir() -> Path:
+    return get_hermes_home() / "skills" / ".curator_backups"
+
+
+def _skills_dir() -> Path:
+    return get_hermes_home() / "skills"
+
+
+def _cron_jobs_file() -> Path:
+    """Source path for the live cron jobs store (``~/.hermes/cron/jobs.json``)."""
+    return get_hermes_home() / "cron" / "jobs.json"
+
+
+CRON_JOBS_FILENAME = "cron-jobs.json"
+
+
+def _backup_cron_jobs_into(dest: Path) -> Dict[str, Any]:
+    """Copy the live cron jobs.json into ``dest`` as ``cron-jobs.json``.
+
+    Returns a small dict describing what was captured so the caller can
+    fold it into the manifest. Never raises — if the cron file is missing
+    or unreadable, the return dict has ``backed_up=False`` and the reason,
+    and the snapshot proceeds without cron data (the snapshot is still
+    useful for rolling back skills).
+    """
+    src = _cron_jobs_file()
+    info: Dict[str, Any] = {"backed_up": False, "jobs_count": 0}
+    if not src.exists():
+        info["reason"] = "no cron/jobs.json present"
+        return info
+    try:
+        raw = src.read_text(encoding="utf-8")
+    except OSError as e:
+        logger.debug("Failed to read cron/jobs.json for backup: %s", e)
+        info["reason"] = f"read error: {e}"
+        return info
+    # Count jobs as a nice diagnostic — but don't fail the snapshot if the
+    # file is unparseable; just store the raw text and let rollback deal
+    # with it (or not, if it's corrupted). jobs.json wraps the list as
+    # `{"jobs": [...], "updated_at": ...}` — we count via that shape, and
+    # fall back to bare-list shape just in case the format ever changes.
+    try:
+        parsed = json.loads(raw)
+        if isinstance(parsed, dict):
+            inner = parsed.get("jobs")
+            if isinstance(inner, list):
+                info["jobs_count"] = len(inner)
+        elif isinstance(parsed, list):
+            info["jobs_count"] = len(parsed)
+    except (json.JSONDecodeError, TypeError):
+        info["jobs_count"] = 0
+        info["parse_warning"] = "jobs.json was not valid JSON at snapshot time"
+    try:
+        (dest / CRON_JOBS_FILENAME).write_text(raw, encoding="utf-8")
+    except OSError as e:
+        logger.debug("Failed to write cron backup file: %s", e)
+        info["reason"] = f"write error: {e}"
+        return info
+    info["backed_up"] = True
+    return info
+
+
+def _utc_id(now: Optional[datetime] = None) -> str:
+    """UTC ISO-ish filesystem-safe timestamp: ``2026-05-01T13-05-42Z``."""
+    if now is None:
+        now = datetime.now(timezone.utc)
+    # isoformat → "2026-05-01T13:05:42.123456+00:00"; strip subseconds and tz.
+    s = now.replace(microsecond=0).isoformat()
+    if s.endswith("+00:00"):
+        s = s[:-6]
+    return s.replace(":", "-") + "Z"
+
+
+def _load_config() -> Dict[str, Any]:
+    try:
+        from hermes_cli.config import load_config
+        cfg = load_config()
+    except Exception as e:
+        logger.debug("Failed to load config for curator backup: %s", e)
+        return {}
+    if not isinstance(cfg, dict):
+        return {}
+    cur = cfg.get("curator") or {}
+    if not isinstance(cur, dict):
+        return {}
+    bk = cur.get("backup") or {}
+    return bk if isinstance(bk, dict) else {}
+
+
+def is_enabled() -> bool:
+    """Default ON — the whole point of the backup is safety by default."""
+    return bool(_load_config().get("enabled", True))
+
+
+def get_keep() -> int:
+    cfg = _load_config()
+    try:
+        n = int(cfg.get("keep", DEFAULT_KEEP))
+    except (TypeError, ValueError):
+        n = DEFAULT_KEEP
+    return max(1, n)
+
+
+# ---------------------------------------------------------------------------
+# Snapshot
+# ---------------------------------------------------------------------------
+
+def _count_skill_files(base: Path) -> int:
+    try:
+        return sum(1 for _ in base.rglob("SKILL.md"))
+    except OSError:
+        return 0
+
+
+def _write_manifest(dest: Path, reason: str, archive_path: Path,
+                    skills_counted: int,
+                    cron_info: Optional[Dict[str, Any]] = None) -> None:
+    manifest = {
+        "id": dest.name,
+        "reason": reason,
+        "created_at": datetime.now(timezone.utc).isoformat(),
+        "archive": archive_path.name,
+        "archive_bytes": archive_path.stat().st_size,
+        "skill_files": skills_counted,
+    }
+    if cron_info is not None:
+        manifest["cron_jobs"] = {
+            "backed_up": bool(cron_info.get("backed_up", False)),
+            "jobs_count": int(cron_info.get("jobs_count", 0)),
+        }
+        if not cron_info.get("backed_up"):
+            manifest["cron_jobs"]["reason"] = cron_info.get("reason", "not captured")
+        if cron_info.get("parse_warning"):
+            manifest["cron_jobs"]["parse_warning"] = cron_info["parse_warning"]
+    (dest / "manifest.json").write_text(
+        json.dumps(manifest, indent=2, sort_keys=True), encoding="utf-8"
+    )
+
+
+def snapshot_skills(reason: str = "manual") -> Optional[Path]:
+    """Create a tar.gz snapshot of ``~/.hermes/skills/`` and prune old ones.
+
+    Returns the snapshot directory path, or ``None`` if the snapshot was
+    skipped (backup disabled, skills dir missing, or an IO error occurred —
+    in which case we log at debug and return None so the curator never
+    aborts a pass because of a backup failure).
+    """
+    if not is_enabled():
+        logger.debug("Curator backup disabled by config; skipping snapshot")
+        return None
+
+    skills = _skills_dir()
+    if not skills.exists():
+        logger.debug("No ~/.hermes/skills/ directory — nothing to back up")
+        return None
+
+    backups = _backups_dir()
+    try:
+        backups.mkdir(parents=True, exist_ok=True)
+    except OSError as e:
+        logger.debug("Failed to create backups dir %s: %s", backups, e)
+        return None
+
+    # Uniquify: if a snapshot with the same second already exists (can
+    # happen if two curator runs fire in the same second), append a short
+    # counter. Avoids clobbering and avoids timestamp collisions.
+    base_id = _utc_id()
+    snap_id = base_id
+    counter = 1
+    while (backups / snap_id).exists():
+        snap_id = f"{base_id}-{counter:02d}"
+        counter += 1
+
+    dest = backups / snap_id
+    try:
+        dest.mkdir(parents=True, exist_ok=False)
+    except OSError as e:
+        logger.debug("Failed to create snapshot dir %s: %s", dest, e)
+        return None
+
+    archive = dest / "skills.tar.gz"
+    try:
+        # Stream into the tarball — no tempdir copy needed.
+        with tarfile.open(archive, "w:gz", compresslevel=6) as tf:
+            for entry in sorted(skills.iterdir()):
+                if entry.name in _EXCLUDE_TOP_LEVEL:
+                    continue
+                # arcname: store paths relative to skills/ so extraction
+                # drops cleanly back into the skills dir.
+                tf.add(str(entry), arcname=entry.name, recursive=True)
+        # Capture cron/jobs.json alongside the tarball. Never fails the
+        # snapshot — the skills side is the core guarantee; cron is
+        # additive. We still record in the manifest whether it was
+        # captured so rollback can surface "no cron data in this snapshot".
+        cron_info = _backup_cron_jobs_into(dest)
+        _write_manifest(dest, reason, archive,
+                        _count_skill_files(skills),
+                        cron_info=cron_info)
+    except (OSError, tarfile.TarError) as e:
+        logger.debug("Curator snapshot failed: %s", e, exc_info=True)
+        # Clean up partial snapshot
+        try:
+            shutil.rmtree(dest, ignore_errors=True)
+        except OSError:
+            pass
+        return None
+
+    _prune_old(keep=get_keep())
+    logger.info("Curator snapshot created: %s (%s)", snap_id, reason)
+    return dest
+
+
+def _prune_old(keep: int) -> List[str]:
+    """Delete regular snapshots beyond the newest *keep*. Returns deleted
+    ids. Staging dirs (``.rollback-staging-*``) are implementation detail
+    and pruned independently on every call."""
+    backups = _backups_dir()
+    if not backups.exists():
+        return []
+    entries: List[Tuple[str, Path]] = []
+    stale_staging: List[Path] = []
+    for child in backups.iterdir():
+        if not child.is_dir():
+            continue
+        if child.name.startswith(".rollback-staging-"):
+            # Staging dirs are only supposed to exist briefly during a
+            # rollback. If we find one here (e.g. from a crashed rollback),
+            # clean it up opportunistically.
+            stale_staging.append(child)
+            continue
+        if _ID_RE.match(child.name):
+            entries.append((child.name, child))
+    # Newest first (lexicographic works because the id is UTC ISO).
+    entries.sort(key=lambda t: t[0], reverse=True)
+    deleted: List[str] = []
+    for _, path in entries[keep:]:
+        try:
+            shutil.rmtree(path)
+            deleted.append(path.name)
+        except OSError as e:
+            logger.debug("Failed to prune %s: %s", path, e)
+    for path in stale_staging:
+        try:
+            shutil.rmtree(path)
+        except OSError as e:
+            logger.debug("Failed to clean stale staging dir %s: %s", path, e)
+    return deleted
+
+
+# ---------------------------------------------------------------------------
+# List + rollback
+# ---------------------------------------------------------------------------
+
+def _read_manifest(snap_dir: Path) -> Dict[str, Any]:
+    mf = snap_dir / "manifest.json"
+    if not mf.exists():
+        return {}
+    try:
+        return json.loads(mf.read_text(encoding="utf-8"))
+    except (OSError, json.JSONDecodeError):
+        return {}
+
+
+def list_backups() -> List[Dict[str, Any]]:
+    """Return all restorable snapshots, newest first. Only entries with a
+    real ``skills.tar.gz`` tarball are listed — transient
+    ``.rollback-staging-*`` directories created mid-rollback are
+    implementation detail and not shown."""
+    backups = _backups_dir()
+    if not backups.exists():
+        return []
+    out: List[Dict[str, Any]] = []
+    for child in sorted(backups.iterdir(), reverse=True):
+        if not child.is_dir():
+            continue
+        if not _ID_RE.match(child.name):
+            continue
+        if not (child / "skills.tar.gz").exists():
+            continue
+        mf = _read_manifest(child)
+        mf.setdefault("id", child.name)
+        mf.setdefault("path", str(child))
+        if "archive_bytes" not in mf:
+            arc = child / "skills.tar.gz"
+            try:
+                mf["archive_bytes"] = arc.stat().st_size
+            except OSError:
+                mf["archive_bytes"] = 0
+        out.append(mf)
+    return out
+
+
+def _resolve_backup(backup_id: Optional[str]) -> Optional[Path]:
+    """Return the path of the requested backup, or the newest one if
+    *backup_id* is None. Returns None if no match."""
+    backups = _backups_dir()
+    if not backups.exists():
+        return None
+    if backup_id:
+        target = backups / backup_id
+        if (
+            target.is_dir()
+            and _ID_RE.match(backup_id)
+            and (target / "skills.tar.gz").exists()
+        ):
+            return target
+        return None
+    candidates = [
+        c for c in sorted(backups.iterdir(), reverse=True)
+        if c.is_dir() and _ID_RE.match(c.name) and (c / "skills.tar.gz").exists()
+    ]
+    return candidates[0] if candidates else None
+
+
+def _restore_cron_skill_links(snapshot_dir: Path) -> Dict[str, Any]:
+    """Reconcile backed-up cron skill links into the live ``cron/jobs.json``.
+
+    We do NOT overwrite the whole cron file. Only the ``skills`` and
+    ``skill`` fields are restored, and only on jobs that still exist in the
+    current file (matched by ``id``). Everything else about the job —
+    schedule, next_run_at, last_run_at, enabled, prompt, workdir, hooks —
+    is live state that the user/scheduler has modified since the snapshot;
+    overwriting it would regress unrelated cron activity.
+
+    Rules:
+    - Jobs present in backup AND live, with differing skills → skills restored.
+    - Jobs present in backup AND live, with matching skills → no-op.
+    - Jobs present in backup but gone from live (user deleted the job
+      after the snapshot) → skipped, noted in the return report.
+    - Jobs present in live but not in backup (user created a new cron
+      job after the snapshot) → left untouched.
+
+    Never raises; failures are captured in the return dict. Writes through
+    ``cron.jobs`` to pick up the same lock + atomic-write path that tick()
+    uses, so we don't race the scheduler.
+    """
+    report: Dict[str, Any] = {
+        "attempted": False,
+        "restored": [],
+        "skipped_missing": [],
+        "unchanged": 0,
+        "error": None,
+    }
+    backup_file = snapshot_dir / CRON_JOBS_FILENAME
+    if not backup_file.exists():
+        report["error"] = f"snapshot has no {CRON_JOBS_FILENAME}"
+        return report
+
+    try:
+        backup_text = backup_file.read_text(encoding="utf-8")
+        backup_parsed = json.loads(backup_text)
+    except (OSError, json.JSONDecodeError) as e:
+        report["error"] = f"failed to load backed-up jobs: {e}"
+        return report
+    # jobs.json on disk is `{"jobs": [...], "updated_at": ...}`; accept both
+    # that shape and a bare list for forward compat.
+    if isinstance(backup_parsed, dict):
+        backup_jobs = backup_parsed.get("jobs")
+    elif isinstance(backup_parsed, list):
+        backup_jobs = backup_parsed
+    else:
+        backup_jobs = None
+    if not isinstance(backup_jobs, list):
+        report["error"] = "backed-up cron-jobs.json has no jobs list"
+        return report
+
+    # Build a lookup of the backed-up skill state keyed by job id.
+    # We only need the two skill-ish fields (legacy single and modern list).
+    backup_by_id: Dict[str, Dict[str, Any]] = {}
+    for job in backup_jobs:
+        if not isinstance(job, dict):
+            continue
+        jid = job.get("id")
+        if not isinstance(jid, str) or not jid:
+            continue
+        backup_by_id[jid] = {
+            "skills": job.get("skills"),
+            "skill": job.get("skill"),
+            "name": job.get("name") or jid,
+        }
+
+    if not backup_by_id:
+        report["attempted"] = True  # we tried but there was nothing to do
+        return report
+
+    # Load and rewrite the live jobs under the scheduler's lock.
+    try:
+        from cron.jobs import load_jobs, save_jobs, _jobs_file_lock
+    except ImportError as e:
+        report["error"] = f"cron module unavailable: {e}"
+        return report
+
+    report["attempted"] = True
+    try:
+        with _jobs_file_lock:
+            live_jobs = load_jobs()
+            changed = False
+
+            live_ids = set()
+            for live in live_jobs:
+                if not isinstance(live, dict):
+                    continue
+                jid = live.get("id")
+                if not isinstance(jid, str) or not jid:
+                    continue
+                live_ids.add(jid)
+
+                backup = backup_by_id.get(jid)
+                if backup is None:
+                    continue  # live job didn't exist at snapshot time
+
+                cur_skills = live.get("skills")
+                cur_skill = live.get("skill")
+                bkp_skills = backup.get("skills")
+                bkp_skill = backup.get("skill")
+
+                if cur_skills == bkp_skills and cur_skill == bkp_skill:
+                    report["unchanged"] += 1
+                    continue
+
+                # Restore. Preserve absence (don't force the key to appear
+                # if the backup didn't have it either).
+                if bkp_skills is None:
+                    live.pop("skills", None)
+                else:
+                    live["skills"] = bkp_skills
+                if bkp_skill is None:
+                    live.pop("skill", None)
+                else:
+                    live["skill"] = bkp_skill
+
+                report["restored"].append({
+                    "job_id": jid,
+                    "job_name": backup.get("name") or jid,
+                    "from": {"skills": cur_skills, "skill": cur_skill},
+                    "to": {"skills": bkp_skills, "skill": bkp_skill},
+                })
+                changed = True
+
+            # Jobs in backup but not in live = user deleted them after snapshot
+            for jid, backup in backup_by_id.items():
+                if jid not in live_ids:
+                    report["skipped_missing"].append({
+                        "job_id": jid,
+                        "job_name": backup.get("name") or jid,
+                    })
+
+            if changed:
+                save_jobs(live_jobs)
+    except Exception as e:  # noqa: BLE001 — rollback must not die mid-restore
+        logger.debug("Cron skill-link restore failed: %s", e, exc_info=True)
+        report["error"] = f"restore failed mid-flight: {e}"
+
+    return report
+
+
+
+def rollback(backup_id: Optional[str] = None) -> Tuple[bool, str, Optional[Path]]:
+    """Restore ``~/.hermes/skills/`` from a snapshot.
+
+    Strategy:
+      1. Resolve the target snapshot (explicit id or newest regular).
+      2. Take a safety snapshot of the CURRENT skills tree under
+         ``.curator_backups/pre-rollback-<ts>/`` so the rollback itself is
+         undoable.
+      3. Move all current top-level entries (except ``.curator_backups``
+         and ``.hub``) into a tempdir.
+      4. Extract the chosen snapshot into ``~/.hermes/skills/``.
+      5. On failure during 4, move the tempdir contents back (best-effort)
+         and return failure.
+
+    Returns ``(ok, message, snapshot_path)``.
+    """
+    target = _resolve_backup(backup_id)
+    if target is None:
+        return (
+            False,
+            f"no matching backup found"
+            + (f" for id '{backup_id}'" if backup_id else "")
+            + " (use `hermes curator rollback --list` to see available snapshots)",
+            None,
+        )
+    archive = target / "skills.tar.gz"
+    if not archive.exists():
+        return (False, f"snapshot {target.name} has no skills.tar.gz — corrupted?", None)
+
+    skills = _skills_dir()
+    skills.mkdir(parents=True, exist_ok=True)
+    backups = _backups_dir()
+    backups.mkdir(parents=True, exist_ok=True)
+
+    # Step 2: safety snapshot of current state FIRST. If this fails we bail
+    # out before touching anything — otherwise a failed extract could leave
+    # the user with no skills.
+    try:
+        snapshot_skills(reason=f"pre-rollback to {target.name}")
+    except Exception as e:
+        return (False, f"pre-rollback safety snapshot failed: {e}", None)
+
+    # Additionally move current entries into an internal staging dir so
+    # the extract happens into an empty skills tree (predictable result).
+    # This dir is implementation detail — not listed as a restorable
+    # backup. The safety snapshot above is the user-facing undo handle.
+    staged = backups / f".rollback-staging-{_utc_id()}"
+    try:
+        staged.mkdir(parents=True, exist_ok=False)
+    except OSError as e:
+        return (False, f"failed to create staging dir: {e}", None)
+
+    moved: List[Tuple[Path, Path]] = []
+    try:
+        for entry in list(skills.iterdir()):
+            if entry.name in _EXCLUDE_TOP_LEVEL:
+                continue
+            dest = staged / entry.name
+            shutil.move(str(entry), str(dest))
+            moved.append((entry, dest))
+    except OSError as e:
+        # Best-effort rollback of the move
+        for orig, dest in moved:
+            try:
+                shutil.move(str(dest), str(orig))
+            except OSError:
+                pass
+        try:
+            shutil.rmtree(staged, ignore_errors=True)
+        except OSError:
+            pass
+        return (False, f"failed to stage current skills: {e}", None)
+
+    # Step 4: extract the snapshot into skills/
+    try:
+        with tarfile.open(archive, "r:gz") as tf:
+            # Python 3.12+ supports filter='data' for safer extraction.
+            # Fall back to the unfiltered call for older interpreters but
+            # still reject absolute paths and .. components defensively.
+            for member in tf.getmembers():
+                name = member.name
+                if name.startswith("/") or ".." in Path(name).parts:
+                    raise tarfile.TarError(
+                        f"refusing to extract unsafe path: {name!r}"
+                    )
+            try:
+                tf.extractall(str(skills), filter="data")  # type: ignore[call-arg]
+            except TypeError:
+                # Python < 3.12 — no filter kwarg
+                tf.extractall(str(skills))
+    except (OSError, tarfile.TarError) as e:
+        # Best-effort recover: move staged contents back
+        for orig, dest in moved:
+            try:
+                shutil.move(str(dest), str(orig))
+            except OSError:
+                pass
+        try:
+            shutil.rmtree(staged, ignore_errors=True)
+        except OSError:
+            pass
+        return (False, f"snapshot extract failed (state restored): {e}", None)
+
+    # Extract succeeded — the staging dir has served its purpose. The
+    # user's undo handle is the safety snapshot tarball we took earlier.
+    try:
+        shutil.rmtree(staged, ignore_errors=True)
+    except OSError:
+        pass
+
+    # Reconcile cron skill-links. Surgical: only the skills/skill fields
+    # on jobs matched by id. Everything else in jobs.json is live state
+    # (schedule, next_run_at, enabled, prompt, etc.) and we leave it
+    # alone. Failures here don't fail the overall rollback — the skills
+    # tree is already restored, which is the main guarantee.
+    cron_report = _restore_cron_skill_links(target)
+
+    summary_bits = [f"restored from snapshot {target.name}"]
+    if cron_report.get("attempted"):
+        restored_n = len(cron_report.get("restored") or [])
+        skipped_n = len(cron_report.get("skipped_missing") or [])
+        if cron_report.get("error"):
+            summary_bits.append(f"cron links: error — {cron_report['error']}")
+        elif restored_n == 0 and skipped_n == 0 and cron_report.get("unchanged", 0) == 0:
+            # Attempted but nothing matched — empty snapshot or no overlapping ids.
+            pass
+        else:
+            parts = []
+            if restored_n:
+                parts.append(f"{restored_n} job(s) had skill links restored")
+            if skipped_n:
+                parts.append(f"{skipped_n} backed-up job(s) no longer exist (skipped)")
+            if cron_report.get("unchanged"):
+                parts.append(f"{cron_report['unchanged']} already matched")
+            summary_bits.append("cron links: " + ", ".join(parts))
+
+    logger.info("Curator rollback: restored from %s (cron_report=%s)",
+                target.name, cron_report)
+    return (True, "; ".join(summary_bits), target)
+
+
+# ---------------------------------------------------------------------------
+# Human-readable summary for CLI
+# ---------------------------------------------------------------------------
+
+def format_size(n: int) -> str:
+    for unit in ("B", "KB", "MB", "GB"):
+        if n < 1024 or unit == "GB":
+            return f"{n:.1f} {unit}" if unit != "B" else f"{n} B"
+        n /= 1024
+    return f"{n:.1f} GB"
+
+
+def summarize_backups() -> str:
+    rows = list_backups()
+    if not rows:
+        return "No curator snapshots yet."
+    lines = [f"{'id':<24}  {'reason':<40}  {'skills':>6}  {'size':>8}"]
+    lines.append("─" * len(lines[0]))
+    for r in rows:
+        lines.append(
+            f"{r.get('id','?'):<24}  "
+            f"{(r.get('reason','?') or '?')[:40]:<40}  "
+            f"{r.get('skill_files', 0):>6}  "
+            f"{format_size(int(r.get('archive_bytes', 0))):>8}"
+        )
+    return "\n".join(lines)

agent/error_classifier.py

@@ -55,6 +55,7 @@ class FailoverReason(enum.Enum):
     thinking_signature = "thinking_signature"  # Anthropic thinking block sig invalid
     long_context_tier = "long_context_tier"    # Anthropic "extra usage" tier gate
     oauth_long_context_beta_forbidden = "oauth_long_context_beta_forbidden"  # Anthropic OAuth subscription rejects 1M context beta — disable beta and retry
+    llama_cpp_grammar_pattern = "llama_cpp_grammar_pattern"  # llama.cpp json-schema-to-grammar rejects regex escapes in `pattern` / `format` — strip from tools and retry
 
     # Catch-all
     unknown = "unknown"                  # Unclassifiable — retry with backoff
@@ -470,6 +471,31 @@ def classify_api_error(
             should_compress=False,
         )
 
+    # llama.cpp's ``json-schema-to-grammar`` converter (used by its OAI
+    # server to build GBNF tool-call parsers) rejects regex escape classes
+    # like ``\d``/``\w``/``\s`` and most ``format`` values. MCP servers
+    # routinely emit ``"pattern": "\\d{4}-\\d{2}-\\d{2}"`` for date/phone/
+    # email params. llama.cpp surfaces this as HTTP 400 with one of a few
+    # recognizable phrases; on match we strip ``pattern``/``format`` from
+    # ``self.tools`` in the retry loop and retry once. Cloud providers are
+    # unaffected — they accept these keywords and we never hit this branch.
+    if (
+        status_code == 400
+        and (
+            "error parsing grammar" in error_msg
+            or "json-schema-to-grammar" in error_msg
+            or (
+                "unable to generate parser" in error_msg
+                and "template" in error_msg
+            )
+        )
+    ):
+        return _result(
+            FailoverReason.llama_cpp_grammar_pattern,
+            retryable=True,
+            should_compress=False,
+        )
+
     # ── 2. HTTP status code classification ──────────────────────────
 
     if status_code is not None:
@@ -520,7 +546,12 @@ def classify_api_error(
 
     is_disconnect = any(p in error_msg for p in _SERVER_DISCONNECT_PATTERNS)
     if is_disconnect and not status_code:
-        is_large = approx_tokens > context_length * 0.6 or approx_tokens > 120000 or num_messages > 200
+        # Absolute token/message-count thresholds are only a proxy for smaller
+        # context windows.  Large-context sessions can have hundreds of
+        # messages while still being far below their actual token budget.
+        is_large = approx_tokens > context_length * 0.6 or (
+            context_length <= 256000 and (approx_tokens > 120000 or num_messages > 200)
+        )
         if is_large:
             return _result(
                 FailoverReason.context_overflow,
@@ -766,7 +797,12 @@ def _classify_400(
         if not err_body_msg:
             err_body_msg = str(body.get("message") or "").strip().lower()
     is_generic = len(err_body_msg) < 30 or err_body_msg in ("error", "")
-    is_large = approx_tokens > context_length * 0.4 or approx_tokens > 80000 or num_messages > 80
+    # Absolute token/message-count thresholds are only a proxy for smaller
+    # context windows.  Large-context sessions can have many messages while
+    # still being far below their actual token budget.
+    is_large = approx_tokens > context_length * 0.4 or (
+        context_length <= 256000 and (approx_tokens > 80000 or num_messages > 80)
+    )
 
     if is_generic and is_large:
         return result_fn(

agent/gemini_native_adapter.py

@@ -679,7 +679,21 @@ def translate_stream_event(event: Dict[str, Any], model: str, tool_call_indices:
     finish_reason_raw = str(cand.get("finishReason") or "")
     if finish_reason_raw:
         mapped = "tool_calls" if tool_call_indices else _map_gemini_finish_reason(finish_reason_raw)
-        chunks.append(_make_stream_chunk(model=model, finish_reason=mapped))
+        finish_chunk = _make_stream_chunk(model=model, finish_reason=mapped)
+        # Attach usage from this event's usageMetadata so the streaming
+        # loop in run_agent.py can record token counts (mirrors the
+        # non-streaming path in translate_gemini_response).
+        usage_meta = event.get("usageMetadata") or {}
+        if usage_meta:
+            finish_chunk.usage = SimpleNamespace(
+                prompt_tokens=int(usage_meta.get("promptTokenCount") or 0),
+                completion_tokens=int(usage_meta.get("candidatesTokenCount") or 0),
+                total_tokens=int(usage_meta.get("totalTokenCount") or 0),
+                prompt_tokens_details=SimpleNamespace(
+                    cached_tokens=int(usage_meta.get("cachedContentTokenCount") or 0),
+                ),
+            )
+        chunks.append(finish_chunk)
     return chunks
 
 

agent/google_oauth.py

@@ -489,16 +489,29 @@ def save_credentials(creds: GoogleCredentials) -> Path:
     """Atomically write creds to disk with 0o600 permissions."""
     path = _credentials_path()
     path.parent.mkdir(parents=True, exist_ok=True)
+    # Tighten parent dir to 0o700 so siblings can't traverse to the creds file.
+    # On Windows this is a no-op (POSIX mode bits aren't enforced); ignore failures.
+    try:
+        os.chmod(path.parent, 0o700)
+    except OSError:
+        pass
     payload = json.dumps(creds.to_dict(), indent=2, sort_keys=True) + "\n"
 
     with _credentials_lock():
         tmp_path = path.with_suffix(f".tmp.{os.getpid()}.{secrets.token_hex(4)}")
         try:
-            with open(tmp_path, "w", encoding="utf-8") as fh:
+            # Create with 0o600 atomically to close the TOCTOU window where the
+            # default umask (often 0o644) would briefly expose tokens to other
+            # local users between open() and chmod().
+            fd = os.open(
+                str(tmp_path),
+                os.O_WRONLY | os.O_CREAT | os.O_EXCL,
+                stat.S_IRUSR | stat.S_IWUSR,
+            )
+            with os.fdopen(fd, "w", encoding="utf-8") as fh:
                 fh.write(payload)
                 fh.flush()
                 os.fsync(fh.fileno())
-            os.chmod(tmp_path, stat.S_IRUSR | stat.S_IWUSR)
             atomic_replace(tmp_path, path)
         finally:
             try:

agent/i18n.py

@@ -0,0 +1,230 @@
+"""Lightweight internationalization (i18n) for Hermes static user-facing messages.
+
+Scope (thin slice, by design): only the highest-impact static strings shown
+to the user by Hermes itself -- approval prompts, a handful of gateway slash
+command replies, restart-drain notices.  Agent-generated output, log lines,
+error tracebacks, tool outputs, and slash-command descriptions all stay in
+English.
+
+Catalog files live under ``locales/<lang>.yaml`` at the repo root.  Each
+catalog is a flat dict keyed by dotted paths (e.g. ``approval.choose`` or
+``gateway.approval_expired``).  Missing keys fall back to English; if English
+is missing too, the key path itself is returned so a broken catalog never
+crashes the agent.
+
+Usage::
+
+    from agent.i18n import t
+    print(t("approval.choose_long"))                       # current lang
+    print(t("gateway.draining", count=3))                  # {count} formatted
+    print(t("approval.choose_long", lang="zh"))            # explicit override
+
+Language resolution order:
+    1. Explicit ``lang=`` argument passed to :func:`t`
+    2. ``HERMES_LANGUAGE`` environment variable (for tests / quick override)
+    3. ``display.language`` from config.yaml
+    4. ``"en"`` (baseline)
+
+Supported languages: en, zh, ja, de, es.  Unknown values fall back to en.
+"""
+
+from __future__ import annotations
+
+import logging
+import os
+import threading
+from functools import lru_cache
+from pathlib import Path
+from typing import Any
+
+logger = logging.getLogger(__name__)
+
+SUPPORTED_LANGUAGES: tuple[str, ...] = ("en", "zh", "ja", "de", "es")
+DEFAULT_LANGUAGE = "en"
+
+# Accept a few natural aliases so users who type "chinese" / "zh-CN" / "jp"
+# get the right catalog instead of silently falling back to English.
+_LANGUAGE_ALIASES: dict[str, str] = {
+    "english": "en", "en-us": "en", "en-gb": "en",
+    "chinese": "zh", "mandarin": "zh", "zh-cn": "zh", "zh-tw": "zh", "zh-hans": "zh", "zh-hant": "zh",
+    "japanese": "ja", "jp": "ja", "ja-jp": "ja",
+    "german": "de", "deutsch": "de", "de-de": "de",
+    "spanish": "es", "español": "es", "espanol": "es", "es-es": "es", "es-mx": "es",
+}
+
+_catalog_cache: dict[str, dict[str, str]] = {}
+_catalog_lock = threading.Lock()
+
+
+def _locales_dir() -> Path:
+    """Return the directory containing locale YAML files.
+
+    Lives next to the repo root so both the bundled install and editable
+    checkouts find it without PYTHONPATH gymnastics.
+    """
+    # agent/i18n.py -> agent/ -> repo root
+    return Path(__file__).resolve().parent.parent / "locales"
+
+
+def _normalize_lang(value: Any) -> str:
+    """Normalize a user-supplied language value to a supported code.
+
+    Accepts supported codes directly, common aliases (``chinese`` -> ``zh``),
+    and case-insensitive regional tags (``zh-CN`` -> ``zh``).  Returns the
+    default language for unknown values.
+    """
+    if not isinstance(value, str):
+        return DEFAULT_LANGUAGE
+    key = value.strip().lower()
+    if not key:
+        return DEFAULT_LANGUAGE
+    if key in SUPPORTED_LANGUAGES:
+        return key
+    if key in _LANGUAGE_ALIASES:
+        return _LANGUAGE_ALIASES[key]
+    # Try stripping a region suffix (e.g. "pt-br" -> "pt" won't be supported,
+    # but "zh-CN" -> "zh" will).
+    base = key.split("-", 1)[0]
+    if base in SUPPORTED_LANGUAGES:
+        return base
+    return DEFAULT_LANGUAGE
+
+
+def _load_catalog(lang: str) -> dict[str, str]:
+    """Load and flatten one locale YAML file into a dotted-key dict.
+
+    YAML files can be nested for human readability; this produces the flat
+    key space :func:`t` expects.  Cached per-language for the process.
+    """
+    with _catalog_lock:
+        cached = _catalog_cache.get(lang)
+        if cached is not None:
+            return cached
+
+    path = _locales_dir() / f"{lang}.yaml"
+    if not path.is_file():
+        logger.debug("i18n catalog missing for %s at %s", lang, path)
+        with _catalog_lock:
+            _catalog_cache[lang] = {}
+        return {}
+
+    try:
+        import yaml  # PyYAML is already a hermes dependency
+        with path.open("r", encoding="utf-8") as f:
+            raw = yaml.safe_load(f) or {}
+    except Exception as exc:
+        logger.warning("Failed to load i18n catalog %s: %s", path, exc)
+        with _catalog_lock:
+            _catalog_cache[lang] = {}
+        return {}
+
+    flat: dict[str, str] = {}
+    _flatten_into(raw, "", flat)
+    with _catalog_lock:
+        _catalog_cache[lang] = flat
+    return flat
+
+
+def _flatten_into(node: Any, prefix: str, out: dict[str, str]) -> None:
+    if isinstance(node, dict):
+        for key, value in node.items():
+            child_key = f"{prefix}.{key}" if prefix else str(key)
+            _flatten_into(value, child_key, out)
+    elif isinstance(node, str):
+        out[prefix] = node
+    # Non-string, non-dict leaves are ignored -- catalogs are text-only.
+
+
+@lru_cache(maxsize=1)
+def _config_language_cached() -> str | None:
+    """Read ``display.language`` from config.yaml once per process.
+
+    Cached because ``t()`` is called in hot paths (every approval prompt,
+    every gateway reply) and re-reading YAML each call would be wasteful.
+    ``reset_language_cache()`` clears this when config changes at runtime
+    (e.g. after the setup wizard).
+    """
+    try:
+        from hermes_cli.config import load_config
+        cfg = load_config()
+        lang = (cfg.get("display") or {}).get("language")
+        if lang:
+            return _normalize_lang(lang)
+    except Exception as exc:
+        logger.debug("Could not read display.language from config: %s", exc)
+    return None
+
+
+def reset_language_cache() -> None:
+    """Invalidate cached language resolution and catalogs.
+
+    Call after :func:`hermes_cli.config.save_config` if a running process
+    needs to pick up a changed ``display.language`` without restart.
+    """
+    _config_language_cached.cache_clear()
+    with _catalog_lock:
+        _catalog_cache.clear()
+
+
+def get_language() -> str:
+    """Resolve the active language using env > config > default order."""
+    env_lang = os.environ.get("HERMES_LANGUAGE")
+    if env_lang:
+        return _normalize_lang(env_lang)
+    cfg_lang = _config_language_cached()
+    if cfg_lang:
+        return cfg_lang
+    return DEFAULT_LANGUAGE
+
+
+def t(key: str, lang: str | None = None, **format_kwargs: Any) -> str:
+    """Translate a dotted key to the active language.
+
+    Parameters
+    ----------
+    key
+        Dotted path into the catalog, e.g. ``"approval.choose_long"``.
+    lang
+        Explicit language override.  Takes precedence over env + config.
+    **format_kwargs
+        ``str.format`` substitution arguments (``t("gateway.drain", count=3)``
+        expects a catalog entry with a ``{count}`` placeholder).
+
+    Returns
+    -------
+    The translated string, or the English fallback if the key is missing in
+    the target language, or the bare key if English is also missing.
+    """
+    target = _normalize_lang(lang) if lang else get_language()
+    catalog = _load_catalog(target)
+    value = catalog.get(key)
+
+    if value is None and target != DEFAULT_LANGUAGE:
+        # Fall through to English rather than showing a key path to the user.
+        value = _load_catalog(DEFAULT_LANGUAGE).get(key)
+
+    if value is None:
+        # Last-ditch: return the key itself.  A broken catalog should not
+        # crash anything; it just looks ugly until someone fixes it.
+        logger.debug("i18n miss: key=%r lang=%r", key, target)
+        value = key
+
+    if format_kwargs:
+        try:
+            return value.format(**format_kwargs)
+        except (KeyError, IndexError, ValueError) as exc:
+            logger.warning(
+                "i18n format failed for key=%r lang=%r kwargs=%r: %s",
+                key, target, format_kwargs, exc,
+            )
+            return value
+    return value
+
+
+__all__ = [
+    "SUPPORTED_LANGUAGES",
+    "DEFAULT_LANGUAGE",
+    "t",
+    "get_language",
+    "reset_language_cache",
+]

agent/manual_compression_feedback.py

@@ -20,25 +20,25 @@ def summarize_manual_compression(
         headline = f"No changes from compression: {before_count} messages"
         if after_tokens == before_tokens:
             token_line = (
-                f"Rough transcript estimate: ~{before_tokens:,} tokens (unchanged)"
+                f"Approx request size: ~{before_tokens:,} tokens (unchanged)"
             )
         else:
             token_line = (
-                f"Rough transcript estimate: ~{before_tokens:,} → "
+                f"Approx request size: ~{before_tokens:,} → "
                 f"~{after_tokens:,} tokens"
             )
     else:
         headline = f"Compressed: {before_count} → {after_count} messages"
         token_line = (
-            f"Rough transcript estimate: ~{before_tokens:,} → "
+            f"Approx request size: ~{before_tokens:,} → "
             f"~{after_tokens:,} tokens"
         )
 
     note = None
     if not noop and after_count < before_count and after_tokens > before_tokens:
         note = (
-            "Note: fewer messages can still raise this rough transcript estimate "
-            "when compression rewrites the transcript into denser summaries."
+            "Note: fewer messages can still raise this estimate when "
+            "compression rewrites the transcript into denser summaries."
         )
 
     return {

agent/moonshot_schema.py

@@ -81,15 +81,56 @@ def _repair_schema(node: Any, is_schema: bool = True) -> Any:
         return repaired
 
     # Rule 2: when anyOf is present, type belongs only on the children.
+    # Additionally, Moonshot rejects null-type branches inside anyOf
+    # (enum value (<nil>) does not match any type in [string]).
+    # Collapse the anyOf to the first non-null branch and infer its type.
     if "anyOf" in repaired and isinstance(repaired["anyOf"], list):
         repaired.pop("type", None)
-        return repaired
+        non_null = [b for b in repaired["anyOf"]
+                    if isinstance(b, dict) and b.get("type") != "null"]
+        if non_null and len(non_null) < len(repaired["anyOf"]):
+            # Drop the anyOf wrapper — keep only the non-null branch.
+            # If there's a single non-null branch, promote it and fall
+            # through to Rules 1/3 so nullable/enum cleanup still applies
+            # to the merged node.
+            if len(non_null) == 1:
+                merge = {k: v for k, v in repaired.items() if k != "anyOf"}
+                merge.update(non_null[0])
+                repaired = merge
+            else:
+                repaired["anyOf"] = non_null
+                return repaired
+        else:
+            # Nothing to collapse — parent type stripped, children already
+            # repaired by the recursive walk above.
+            return repaired
+
+    # Moonshot also rejects non-standard keywords like ``nullable`` on
+    # parameter schemas — strip it.
+    repaired.pop("nullable", None)
 
     # Rule 1: property schemas without type need one.  $ref nodes are exempt
     # — their type comes from the referenced definition.
-    if "$ref" in repaired:
-        return repaired
-    return _fill_missing_type(repaired)
+    # Fill missing type BEFORE Rule 3 so enum cleanup can check the type.
+    if "$ref" not in repaired:
+        repaired = _fill_missing_type(repaired)
+
+    # Rule 3: Moonshot rejects null/empty-string values inside enum arrays
+    # when the parent type is a scalar (string, integer, etc.).  The error:
+    #   "enum value (<nil>) does not match any type in [string]"
+    # Strip null and empty-string from enum values, and if the enum becomes
+    # empty, drop it entirely.
+    if "enum" in repaired and isinstance(repaired["enum"], list):
+        node_type = repaired.get("type")
+        if node_type in ("string", "integer", "number", "boolean"):
+            cleaned = [v for v in repaired["enum"]
+                       if v is not None and v != ""]
+            if cleaned:
+                repaired["enum"] = cleaned
+            else:
+                repaired.pop("enum")
+
+    return repaired
 
 
 def _fill_missing_type(node: Dict[str, Any]) -> Dict[str, Any]:

agent/prompt_builder.py

@@ -183,8 +183,8 @@ SKILLS_GUIDANCE = (
 )
 
 KANBAN_GUIDANCE = (
-    "# You are a Kanban worker\n"
-    "You were spawned by the Hermes Kanban dispatcher to execute ONE task from "
+    "# Kanban task execution protocol\n"
+    "You have been assigned ONE task from "
     "the shared board at `~/.hermes/kanban.db`. Your task id is in "
     "`$HERMES_KANBAN_TASK`; your workspace is `$HERMES_KANBAN_WORKSPACE`. "
     "The `kanban_*` tools in your schema are your primary coordination surface — "
@@ -513,6 +513,12 @@ PLATFORM_HINTS = {
         "image and is the WRONG path. Bare Unicode emoji in text is also not a substitute "
         "— when a sticker is the right response, use yb_send_sticker."
     ),
+    "api_server": (
+        "You're responding through an API server. The rendering layer is unknown — "
+        "assume plain text. No markdown formatting (no asterisks, bullets, headers, "
+        "code fences). Treat this like a conversation, not a document. Keep responses "
+        "brief and natural."
+    ),
 }
 
 # ---------------------------------------------------------------------------

agent/redact.py

@@ -305,13 +305,18 @@ def _redact_form_body(text: str) -> str:
     return _redact_query_string(text.strip())
 
 
-def redact_sensitive_text(text: str, *, force: bool = False) -> str:
+def redact_sensitive_text(text: str, *, force: bool = False, code_file: bool = False) -> str:
     """Apply all redaction patterns to a block of text.
 
     Safe to call on any string -- non-matching text passes through unchanged.
     Disabled by default — enable via security.redact_secrets: true in config.yaml.
     Set force=True for safety boundaries that must never return raw secrets
     regardless of the user's global logging redaction preference.
+
+    Set code_file=True to skip the ENV-assignment and JSON-field regex
+    patterns when the text is known to be source code (e.g. MAX_TOKENS=***
+    constants, "apiKey": "test" fixtures). Prefix patterns, auth headers,
+    private keys, DB connstrings, JWTs, and URL secrets are still redacted.
     """
     if text is None:
         return None
@@ -325,17 +330,18 @@ def redact_sensitive_text(text: str, *, force: bool = False) -> str:
     # Known prefixes (sk-, ghp_, etc.)
     text = _PREFIX_RE.sub(lambda m: _mask_token(m.group(1)), text)
 
-    # ENV assignments: OPENAI_API_KEY=sk-abc...
-    def _redact_env(m):
-        name, quote, value = m.group(1), m.group(2), m.group(3)
-        return f"{name}={quote}{_mask_token(value)}{quote}"
-    text = _ENV_ASSIGN_RE.sub(_redact_env, text)
+    # ENV assignments: OPENAI_API_KEY=***  (skip for code files — false positives)
+    if not code_file:
+        def _redact_env(m):
+            name, quote, value = m.group(1), m.group(2), m.group(3)
+            return f"{name}={quote}{_mask_token(value)}{quote}"
+        text = _ENV_ASSIGN_RE.sub(_redact_env, text)
 
-    # JSON fields: "apiKey": "value"
-    def _redact_json(m):
-        key, value = m.group(1), m.group(2)
-        return f'{key}: "{_mask_token(value)}"'
-    text = _JSON_FIELD_RE.sub(_redact_json, text)
+        # JSON fields: "apiKey": "***"  (skip for code files — false positives)
+        def _redact_json(m):
+            key, value = m.group(1), m.group(2)
+            return f'{key}: "{_mask_token(value)}"'
+        text = _JSON_FIELD_RE.sub(_redact_json, text)
 
     # Authorization headers
     text = _AUTH_HEADER_RE.sub(

agent/skill_commands.py

@@ -6,6 +6,7 @@ can invoke skills via /skill-name commands.
 
 import json
 import logging
+import os
 import re
 from pathlib import Path
 from typing import Any, Dict, Optional
@@ -20,10 +21,35 @@ from agent.skill_preprocessing import (
 logger = logging.getLogger(__name__)
 
 _skill_commands: Dict[str, Dict[str, Any]] = {}
+_skill_commands_platform: Optional[str] = None
 # Patterns for sanitizing skill names into clean hyphen-separated slugs.
 _SKILL_INVALID_CHARS = re.compile(r"[^a-z0-9-]")
 _SKILL_MULTI_HYPHEN = re.compile(r"-{2,}")
 
+
+def _resolve_skill_commands_platform() -> Optional[str]:
+    """Return the current platform scope used for disabled-skill filtering.
+
+    Used to detect when the active platform has shifted so
+    :func:`get_skill_commands` can drop a stale cache that was populated
+    for a different platform's ``skills.platform_disabled`` view (#14536).
+
+    Resolves from (in order) ``HERMES_PLATFORM`` env var and
+    ``HERMES_SESSION_PLATFORM`` from the gateway session context. Returns
+    ``None`` when no platform scope is active (e.g. classic CLI, RL
+    rollouts, standalone scripts).
+    """
+    try:
+        from gateway.session_context import get_session_env
+
+        resolved_platform = (
+            os.getenv("HERMES_PLATFORM")
+            or get_session_env("HERMES_SESSION_PLATFORM")
+        )
+    except Exception:
+        resolved_platform = os.getenv("HERMES_PLATFORM")
+    return resolved_platform or None
+
 def _load_skill_payload(skill_identifier: str, task_id: str | None = None) -> tuple[dict[str, Any], Path | None, str] | None:
     """Load a skill by name/path and return (loaded_payload, skill_dir, display_name)."""
     raw_identifier = (skill_identifier or "").strip()
@@ -218,7 +244,8 @@ def scan_skill_commands() -> Dict[str, Dict[str, Any]]:
     Returns:
         Dict mapping "/skill-name" to {name, description, skill_md_path, skill_dir}.
     """
-    global _skill_commands
+    global _skill_commands, _skill_commands_platform
+    _skill_commands_platform = _resolve_skill_commands_platform()
     _skill_commands = {}
     try:
         from tools.skills_tool import SKILLS_DIR, _parse_frontmatter, skill_matches_platform, _get_disabled_skill_names
@@ -278,8 +305,16 @@ def scan_skill_commands() -> Dict[str, Dict[str, Any]]:
 
 
 def get_skill_commands() -> Dict[str, Dict[str, Any]]:
-    """Return the current skill commands mapping (scan first if empty)."""
-    if not _skill_commands:
+    """Return the current skill commands mapping (scan first if empty).
+
+    Rescans when the active platform scope changes (e.g. a gateway
+    process serving Telegram and Discord concurrently) so each platform
+    sees its own ``skills.platform_disabled`` view (#14536).
+    """
+    if (
+        not _skill_commands
+        or _skill_commands_platform != _resolve_skill_commands_platform()
+    ):
         scan_skill_commands()
     return _skill_commands
 

agent/think_scrubber.py

@@ -0,0 +1,386 @@
+"""Stateful scrubber for reasoning/thinking blocks in streamed assistant text.
+
+``run_agent._strip_think_blocks`` is regex-based and correct for a complete
+string, but when it runs *per-delta* in ``_fire_stream_delta`` it destroys
+the state that downstream consumers (CLI ``_stream_delta``, gateway
+``GatewayStreamConsumer._filter_and_accumulate``) rely on.
+
+Concretely, when MiniMax-M2.7 streams
+
+    delta1 = "<think>"
+    delta2 = "Let me check their config"
+    delta3 = "</think>"
+
+the per-delta regex erases delta1 entirely (case 2: unterminated-open at
+boundary matches ``^<think>...``), so the downstream state machine never
+sees the open tag, treats delta2 as regular content, and leaks reasoning
+to the user.  Consumers that don't run their own state machine (ACP,
+api_server, TTS) never had any defence at all — they just emitted
+whatever survived the upstream regex.
+
+This module centralises the tag-suppression state machine at the
+upstream layer so every stream_delta_callback sees text that has
+already had reasoning blocks removed.  Partial tags at delta
+boundaries are held back until the next delta resolves them, and
+end-of-stream flushing surfaces any held-back prose that turned out
+not to be a real tag.
+
+Usage::
+
+    scrubber = StreamingThinkScrubber()
+    for delta in stream:
+        visible = scrubber.feed(delta)
+        if visible:
+            emit(visible)
+    tail = scrubber.flush()  # at end of stream
+    if tail:
+        emit(tail)
+
+The scrubber is re-entrant per agent instance.  Call ``reset()`` at
+the top of each new turn so a hung block from an interrupted prior
+stream cannot taint the next turn's output.
+
+Tag variants handled (case-insensitive):
+  ``<think>``, ``<thinking>``, ``<reasoning>``, ``<thought>``,
+  ``<REASONING_SCRATCHPAD>``.
+
+Block-boundary rule for opens: an opening tag is only treated as a
+reasoning-block opener when it appears at the start of the stream,
+after a newline (optionally followed by whitespace), or when only
+whitespace has been emitted on the current line.  This prevents prose
+that *mentions* the tag name (e.g. ``"use <think> tags here"``) from
+being incorrectly suppressed.  Closed pairs (``<think>X</think>``) are
+always suppressed regardless of boundary; a closed pair is an
+intentional, bounded construct.
+"""
+
+from __future__ import annotations
+
+from typing import Tuple
+
+__all__ = ["StreamingThinkScrubber"]
+
+
+class StreamingThinkScrubber:
+    """Stateful scrubber for streaming reasoning/thinking blocks.
+
+    State machine:
+      - ``_in_block``: True while inside an opened block, waiting for
+        a close tag.  All text inside is discarded.
+      - ``_buf``: held-back partial-tag tail.  Emitted / discarded on
+        the next ``feed()`` call or by ``flush()``.
+      - ``_last_emitted_ended_newline``: True iff the most recent
+        emission to the consumer ended with ``\\n``, or nothing has
+        been emitted yet (start-of-stream counts as a boundary).  Used
+        to decide whether an open tag at buffer position 0 is at a
+        block boundary.
+    """
+
+    _OPEN_TAG_NAMES: Tuple[str, ...] = (
+        "think",
+        "thinking",
+        "reasoning",
+        "thought",
+        "REASONING_SCRATCHPAD",
+    )
+
+    # Materialise literal tag strings so the hot path does string
+    # operations, not regex compilation per feed().
+    _OPEN_TAGS: Tuple[str, ...] = tuple(f"<{name}>" for name in _OPEN_TAG_NAMES)
+    _CLOSE_TAGS: Tuple[str, ...] = tuple(f"</{name}>" for name in _OPEN_TAG_NAMES)
+
+    # Pre-compute the longest tag (for partial-tag hold-back bound).
+    _MAX_TAG_LEN: int = max(len(tag) for tag in _OPEN_TAGS + _CLOSE_TAGS)
+
+    def __init__(self) -> None:
+        self._in_block: bool = False
+        self._buf: str = ""
+        self._last_emitted_ended_newline: bool = True
+
+    def reset(self) -> None:
+        """Reset all state.  Call at the top of every new turn."""
+        self._in_block = False
+        self._buf = ""
+        self._last_emitted_ended_newline = True
+
+    def feed(self, text: str) -> str:
+        """Feed one delta; return the scrubbed visible portion.
+
+        May return an empty string when the entire delta is reasoning
+        content or is being held back pending resolution of a partial
+        tag at the boundary.
+        """
+        if not text:
+            return ""
+        buf = self._buf + text
+        self._buf = ""
+        out: list[str] = []
+
+        while buf:
+            if self._in_block:
+                # Hunt for the earliest close tag.
+                close_idx, close_len = self._find_first_tag(
+                    buf, self._CLOSE_TAGS,
+                )
+                if close_idx == -1:
+                    # No close yet — hold back a potential partial
+                    # close-tag prefix; discard everything else.
+                    held = self._max_partial_suffix(buf, self._CLOSE_TAGS)
+                    self._buf = buf[-held:] if held else ""
+                    return "".join(out)
+                # Found close: discard block content + tag, continue.
+                buf = buf[close_idx + close_len:]
+                self._in_block = False
+            else:
+                # Priority 1 — closed <tag>X</tag> pair anywhere in
+                # buf.  Closed pairs are always an intentional,
+                # bounded construct (even mid-line prose containing
+                # an open/close pair is almost certainly a model
+                # leaking reasoning inline), so no boundary gating.
+                pair = self._find_earliest_closed_pair(buf)
+                # Priority 2 — unterminated open tag at a block
+                # boundary.  Boundary-gated so prose that mentions
+                # '<think>' isn't over-stripped.
+                open_idx, open_len = self._find_open_at_boundary(
+                    buf, out,
+                )
+
+                # Pick whichever match comes earliest in the buffer.
+                if pair is not None and (
+                    open_idx == -1 or pair[0] <= open_idx
+                ):
+                    start_idx, end_idx = pair
+                    preceding = buf[:start_idx]
+                    if preceding:
+                        preceding = self._strip_orphan_close_tags(preceding)
+                        if preceding:
+                            out.append(preceding)
+                            self._last_emitted_ended_newline = (
+                                preceding.endswith("\n")
+                            )
+                    buf = buf[end_idx:]
+                    continue
+
+                if open_idx != -1:
+                    # Unterminated open at boundary — emit preceding,
+                    # enter block, continue loop with remainder.
+                    preceding = buf[:open_idx]
+                    if preceding:
+                        preceding = self._strip_orphan_close_tags(preceding)
+                        if preceding:
+                            out.append(preceding)
+                            self._last_emitted_ended_newline = (
+                                preceding.endswith("\n")
+                            )
+                    self._in_block = True
+                    buf = buf[open_idx + open_len:]
+                    continue
+
+                # No resolvable tag structure in buf.  Hold back any
+                # partial-tag prefix at the tail so a split tag
+                # across deltas isn't missed, then emit the rest.
+                held = self._max_partial_suffix(buf, self._OPEN_TAGS)
+                held_close = self._max_partial_suffix(
+                    buf, self._CLOSE_TAGS,
+                )
+                held = max(held, held_close)
+                if held:
+                    emit_text = buf[:-held]
+                    self._buf = buf[-held:]
+                else:
+                    emit_text = buf
+                    self._buf = ""
+                if emit_text:
+                    emit_text = self._strip_orphan_close_tags(emit_text)
+                    if emit_text:
+                        out.append(emit_text)
+                        self._last_emitted_ended_newline = (
+                            emit_text.endswith("\n")
+                        )
+                return "".join(out)
+
+        return "".join(out)
+
+    def flush(self) -> str:
+        """End-of-stream flush.
+
+        If still inside an unterminated block, held-back content is
+        discarded — leaking partial reasoning is worse than a
+        truncated answer.  Otherwise the held-back partial-tag tail is
+        emitted verbatim (it turned out not to be a real tag prefix).
+        """
+        if self._in_block:
+            self._buf = ""
+            self._in_block = False
+            return ""
+        tail = self._buf
+        self._buf = ""
+        if not tail:
+            return ""
+        tail = self._strip_orphan_close_tags(tail)
+        if tail:
+            self._last_emitted_ended_newline = tail.endswith("\n")
+        return tail
+
+    # ── internal helpers ───────────────────────────────────────────────
+
+    @staticmethod
+    def _find_first_tag(
+        buf: str, tags: Tuple[str, ...],
+    ) -> Tuple[int, int]:
+        """Return (earliest_index, tag_length) over *tags*, or (-1, 0).
+
+        Case-insensitive match.
+        """
+        buf_lower = buf.lower()
+        best_idx = -1
+        best_len = 0
+        for tag in tags:
+            idx = buf_lower.find(tag.lower())
+            if idx != -1 and (best_idx == -1 or idx < best_idx):
+                best_idx = idx
+                best_len = len(tag)
+        return best_idx, best_len
+
+    def _find_earliest_closed_pair(self, buf: str):
+        """Return (start_idx, end_idx) of the earliest closed pair, else None.
+
+        A closed pair is ``<tag>...</tag>`` of any variant.  Matches are
+        case-insensitive and non-greedy (the closest close tag after
+        an open tag wins), matching the regex ``<tag>.*?</tag>``
+        semantics of ``_strip_think_blocks`` case 1.  When two tag
+        variants could both match, the one whose open tag appears
+        earlier wins.
+        """
+        buf_lower = buf.lower()
+        best: "tuple[int, int] | None" = None
+        for open_tag, close_tag in zip(self._OPEN_TAGS, self._CLOSE_TAGS):
+            open_lower = open_tag.lower()
+            close_lower = close_tag.lower()
+            open_idx = buf_lower.find(open_lower)
+            if open_idx == -1:
+                continue
+            close_idx = buf_lower.find(
+                close_lower, open_idx + len(open_lower),
+            )
+            if close_idx == -1:
+                continue
+            end_idx = close_idx + len(close_lower)
+            if best is None or open_idx < best[0]:
+                best = (open_idx, end_idx)
+        return best
+
+    def _find_open_at_boundary(
+        self, buf: str, already_emitted: list[str],
+    ) -> Tuple[int, int]:
+        """Return the earliest block-boundary open-tag (idx, len).
+
+        Returns (-1, 0) if no boundary-legal opener is present.
+        """
+        buf_lower = buf.lower()
+        best_idx = -1
+        best_len = 0
+        for tag in self._OPEN_TAGS:
+            tag_lower = tag.lower()
+            search_start = 0
+            while True:
+                idx = buf_lower.find(tag_lower, search_start)
+                if idx == -1:
+                    break
+                if self._is_block_boundary(buf, idx, already_emitted):
+                    if best_idx == -1 or idx < best_idx:
+                        best_idx = idx
+                        best_len = len(tag)
+                    break  # first boundary hit for this tag is enough
+                search_start = idx + 1
+        return best_idx, best_len
+
+    def _is_block_boundary(
+        self, buf: str, idx: int, already_emitted: list[str],
+    ) -> bool:
+        """True iff position *idx* in *buf* is a block boundary.
+
+        A block boundary is:
+          - buf position 0 AND the most recent emission ended with
+            a newline (or nothing has been emitted yet)
+          - any position whose preceding text on the current line
+            (since the last newline in buf) is whitespace-only, AND
+            if there is no newline in the preceding buf portion, the
+            most recent prior emission ended with a newline
+        """
+        if idx == 0:
+            # Check whether the last already-emitted chunk in THIS
+            # feed() call ended with a newline, otherwise fall back
+            # to the cross-feed flag.
+            if already_emitted:
+                return already_emitted[-1].endswith("\n")
+            return self._last_emitted_ended_newline
+        preceding = buf[:idx]
+        last_nl = preceding.rfind("\n")
+        if last_nl == -1:
+            # No newline in buf before the tag — boundary only if the
+            # prior emission ended with a newline AND everything since
+            # is whitespace.
+            if already_emitted:
+                prior_newline = already_emitted[-1].endswith("\n")
+            else:
+                prior_newline = self._last_emitted_ended_newline
+            return prior_newline and preceding.strip() == ""
+        # Newline present — text between it and the tag must be
+        # whitespace-only.
+        return preceding[last_nl + 1:].strip() == ""
+
+    @classmethod
+    def _max_partial_suffix(
+        cls, buf: str, tags: Tuple[str, ...],
+    ) -> int:
+        """Return the longest buf-suffix that is a prefix of any tag.
+
+        Only prefixes strictly shorter than the tag itself count
+        (full-length suffixes are the tag and are handled as matches,
+        not held-back partials).  Case-insensitive.
+        """
+        if not buf:
+            return 0
+        buf_lower = buf.lower()
+        max_check = min(len(buf_lower), cls._MAX_TAG_LEN - 1)
+        for i in range(max_check, 0, -1):
+            suffix = buf_lower[-i:]
+            for tag in tags:
+                tag_lower = tag.lower()
+                if len(tag_lower) > i and tag_lower.startswith(suffix):
+                    return i
+        return 0
+
+    @classmethod
+    def _strip_orphan_close_tags(cls, text: str) -> str:
+        """Remove any close tags from *text* (orphan-close handling).
+
+        An orphan close tag has no matching open in the current
+        scrubber state; it's always noise, stripped with any trailing
+        whitespace so the surrounding prose flows naturally.
+        """
+        if "</" not in text:
+            return text
+        text_lower = text.lower()
+        out: list[str] = []
+        i = 0
+        while i < len(text):
+            matched = False
+            if text_lower[i:i + 2] == "</":
+                for tag in cls._CLOSE_TAGS:
+                    tag_lower = tag.lower()
+                    tag_len = len(tag_lower)
+                    if text_lower[i:i + tag_len] == tag_lower:
+                        # Skip the tag and any trailing whitespace,
+                        # matching _strip_think_blocks case 3.
+                        j = i + tag_len
+                        while j < len(text) and text[j] in " \t\n\r":
+                            j += 1
+                        i = j
+                        matched = True
+                        break
+            if not matched:
+                out.append(text[i])
+                i += 1
+        return "".join(out)

agent/title_generator.py

@@ -17,6 +17,7 @@ logger = logging.getLogger(__name__)
 # so silent-drops (e.g. OpenRouter 402 exhausting the fallback chain)
 # become visible instead of piling up as NULL session titles.
 FailureCallback = Callable[[str, BaseException], None]
+TitleCallback = Callable[[str], None]
 
 _TITLE_PROMPT = (
     "Generate a short, descriptive title (3-7 words) for a conversation that starts with the "
@@ -90,6 +91,7 @@ def auto_title_session(
     assistant_response: str,
     failure_callback: Optional[FailureCallback] = None,
     main_runtime: dict = None,
+    title_callback: Optional[TitleCallback] = None,
 ) -> None:
     """Generate and set a session title if one doesn't already exist.
 
@@ -119,6 +121,11 @@ def auto_title_session(
     try:
         session_db.set_session_title(session_id, title)
         logger.debug("Auto-generated session title: %s", title)
+        if title_callback is not None:
+            try:
+                title_callback(title)
+            except Exception:
+                logger.debug("Auto-title callback failed", exc_info=True)
     except Exception as e:
         logger.debug("Failed to set auto-generated title: %s", e)
 
@@ -131,6 +138,7 @@ def maybe_auto_title(
     conversation_history: list,
     failure_callback: Optional[FailureCallback] = None,
     main_runtime: dict = None,
+    title_callback: Optional[TitleCallback] = None,
 ) -> None:
     """Fire-and-forget title generation after the first exchange.
 
@@ -152,7 +160,11 @@ def maybe_auto_title(
     thread = threading.Thread(
         target=auto_title_session,
         args=(session_db, session_id, user_message, assistant_response),
-        kwargs={"failure_callback": failure_callback, "main_runtime": main_runtime},
+        kwargs={
+            "failure_callback": failure_callback,
+            "main_runtime": main_runtime,
+            "title_callback": title_callback,
+        },
         daemon=True,
         name="auto-title",
     )

agent/tool_guardrails.py

@@ -0,0 +1,455 @@
+"""Pure tool-call loop guardrail primitives.
+
+The controller in this module is intentionally side-effect free: it tracks
+per-turn tool-call observations and returns decisions. Runtime code owns whether
+those decisions become warning guidance, synthetic tool results, or controlled
+turn halts.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+from dataclasses import dataclass, field
+from typing import Any, Mapping
+
+from utils import safe_json_loads
+
+
+IDEMPOTENT_TOOL_NAMES = frozenset(
+    {
+        "read_file",
+        "search_files",
+        "web_search",
+        "web_extract",
+        "session_search",
+        "browser_snapshot",
+        "browser_console",
+        "browser_get_images",
+        "mcp_filesystem_read_file",
+        "mcp_filesystem_read_text_file",
+        "mcp_filesystem_read_multiple_files",
+        "mcp_filesystem_list_directory",
+        "mcp_filesystem_list_directory_with_sizes",
+        "mcp_filesystem_directory_tree",
+        "mcp_filesystem_get_file_info",
+        "mcp_filesystem_search_files",
+    }
+)
+
+MUTATING_TOOL_NAMES = frozenset(
+    {
+        "terminal",
+        "execute_code",
+        "write_file",
+        "patch",
+        "todo",
+        "memory",
+        "skill_manage",
+        "browser_click",
+        "browser_type",
+        "browser_press",
+        "browser_scroll",
+        "browser_navigate",
+        "send_message",
+        "cronjob",
+        "delegate_task",
+        "process",
+    }
+)
+
+
+@dataclass(frozen=True)
+class ToolCallGuardrailConfig:
+    """Thresholds for per-turn tool-call loop detection.
+
+    Warnings are enabled by default and never prevent tool execution. Hard stops
+    are explicit opt-in so interactive CLI/TUI sessions get a gentle nudge unless
+    the user enables circuit-breaker behavior in config.yaml.
+    """
+
+    warnings_enabled: bool = True
+    hard_stop_enabled: bool = False
+    exact_failure_warn_after: int = 2
+    exact_failure_block_after: int = 5
+    same_tool_failure_warn_after: int = 3
+    same_tool_failure_halt_after: int = 8
+    no_progress_warn_after: int = 2
+    no_progress_block_after: int = 5
+    idempotent_tools: frozenset[str] = field(default_factory=lambda: IDEMPOTENT_TOOL_NAMES)
+    mutating_tools: frozenset[str] = field(default_factory=lambda: MUTATING_TOOL_NAMES)
+
+    @classmethod
+    def from_mapping(cls, data: Mapping[str, Any] | None) -> "ToolCallGuardrailConfig":
+        """Build config from the `tool_loop_guardrails` config.yaml section."""
+        if not isinstance(data, Mapping):
+            return cls()
+
+        warn_after = data.get("warn_after")
+        if not isinstance(warn_after, Mapping):
+            warn_after = {}
+        hard_stop_after = data.get("hard_stop_after")
+        if not isinstance(hard_stop_after, Mapping):
+            hard_stop_after = {}
+
+        defaults = cls()
+        return cls(
+            warnings_enabled=_as_bool(data.get("warnings_enabled"), defaults.warnings_enabled),
+            hard_stop_enabled=_as_bool(data.get("hard_stop_enabled"), defaults.hard_stop_enabled),
+            exact_failure_warn_after=_positive_int(
+                warn_after.get("exact_failure", data.get("exact_failure_warn_after")),
+                defaults.exact_failure_warn_after,
+            ),
+            same_tool_failure_warn_after=_positive_int(
+                warn_after.get("same_tool_failure", data.get("same_tool_failure_warn_after")),
+                defaults.same_tool_failure_warn_after,
+            ),
+            no_progress_warn_after=_positive_int(
+                warn_after.get("idempotent_no_progress", data.get("no_progress_warn_after")),
+                defaults.no_progress_warn_after,
+            ),
+            exact_failure_block_after=_positive_int(
+                hard_stop_after.get("exact_failure", data.get("exact_failure_block_after")),
+                defaults.exact_failure_block_after,
+            ),
+            same_tool_failure_halt_after=_positive_int(
+                hard_stop_after.get("same_tool_failure", data.get("same_tool_failure_halt_after")),
+                defaults.same_tool_failure_halt_after,
+            ),
+            no_progress_block_after=_positive_int(
+                hard_stop_after.get("idempotent_no_progress", data.get("no_progress_block_after")),
+                defaults.no_progress_block_after,
+            ),
+        )
+
+
+@dataclass(frozen=True)
+class ToolCallSignature:
+    """Stable, non-reversible identity for a tool name plus canonical args."""
+
+    tool_name: str
+    args_hash: str
+
+    @classmethod
+    def from_call(cls, tool_name: str, args: Mapping[str, Any] | None) -> "ToolCallSignature":
+        canonical = canonical_tool_args(args or {})
+        return cls(tool_name=tool_name, args_hash=_sha256(canonical))
+
+    def to_metadata(self) -> dict[str, str]:
+        """Return public metadata without raw argument values."""
+        return {"tool_name": self.tool_name, "args_hash": self.args_hash}
+
+
+@dataclass(frozen=True)
+class ToolGuardrailDecision:
+    """Decision returned by the tool-call guardrail controller."""
+
+    action: str = "allow"  # allow | warn | block | halt
+    code: str = "allow"
+    message: str = ""
+    tool_name: str = ""
+    count: int = 0
+    signature: ToolCallSignature | None = None
+
+    @property
+    def allows_execution(self) -> bool:
+        return self.action in {"allow", "warn"}
+
+    @property
+    def should_halt(self) -> bool:
+        return self.action in {"block", "halt"}
+
+    def to_metadata(self) -> dict[str, Any]:
+        data: dict[str, Any] = {
+            "action": self.action,
+            "code": self.code,
+            "message": self.message,
+            "tool_name": self.tool_name,
+            "count": self.count,
+        }
+        if self.signature is not None:
+            data["signature"] = self.signature.to_metadata()
+        return data
+
+
+def canonical_tool_args(args: Mapping[str, Any]) -> str:
+    """Return sorted compact JSON for parsed tool arguments."""
+    if not isinstance(args, Mapping):
+        raise TypeError(f"tool args must be a mapping, got {type(args).__name__}")
+    return json.dumps(
+        args,
+        ensure_ascii=False,
+        sort_keys=True,
+        separators=(",", ":"),
+        default=str,
+    )
+
+
+def classify_tool_failure(tool_name: str, result: str | None) -> tuple[bool, str]:
+    """Safety-fallback classifier used only when callers don't pass ``failed``.
+
+    Mirrors ``agent.display._detect_tool_failure`` exactly so the guardrail
+    never disagrees with the CLI's user-visible ``[error]`` tag. Production
+    callers in ``run_agent.py`` always pass an explicit ``failed=`` derived
+    from ``_detect_tool_failure``; this function exists so standalone callers
+    (tests, tooling) still get consistent behavior.
+    """
+    if result is None:
+        return False, ""
+
+    if tool_name == "terminal":
+        data = safe_json_loads(result)
+        if isinstance(data, dict):
+            exit_code = data.get("exit_code")
+            if exit_code is not None and exit_code != 0:
+                return True, f" [exit {exit_code}]"
+        return False, ""
+
+    if tool_name == "memory":
+        data = safe_json_loads(result)
+        if isinstance(data, dict):
+            if data.get("success") is False and "exceed the limit" in data.get("error", ""):
+                return True, " [full]"
+
+    lower = result[:500].lower()
+    if '"error"' in lower or '"failed"' in lower or result.startswith("Error"):
+        return True, " [error]"
+
+    return False, ""
+
+
+class ToolCallGuardrailController:
+    """Per-turn controller for repeated failed/non-progressing tool calls."""
+
+    def __init__(self, config: ToolCallGuardrailConfig | None = None):
+        self.config = config or ToolCallGuardrailConfig()
+        self.reset_for_turn()
+
+    def reset_for_turn(self) -> None:
+        self._exact_failure_counts: dict[ToolCallSignature, int] = {}
+        self._same_tool_failure_counts: dict[str, int] = {}
+        self._no_progress: dict[ToolCallSignature, tuple[str, int]] = {}
+        self._halt_decision: ToolGuardrailDecision | None = None
+
+    @property
+    def halt_decision(self) -> ToolGuardrailDecision | None:
+        return self._halt_decision
+
+    def before_call(self, tool_name: str, args: Mapping[str, Any] | None) -> ToolGuardrailDecision:
+        signature = ToolCallSignature.from_call(tool_name, _coerce_args(args))
+        if not self.config.hard_stop_enabled:
+            return ToolGuardrailDecision(tool_name=tool_name, signature=signature)
+
+        exact_count = self._exact_failure_counts.get(signature, 0)
+        if exact_count >= self.config.exact_failure_block_after:
+            decision = ToolGuardrailDecision(
+                action="block",
+                code="repeated_exact_failure_block",
+                message=(
+                    f"Blocked {tool_name}: the same tool call failed {exact_count} "
+                    "times with identical arguments. Stop retrying it unchanged; "
+                    "change strategy or explain the blocker."
+                ),
+                tool_name=tool_name,
+                count=exact_count,
+                signature=signature,
+            )
+            self._halt_decision = decision
+            return decision
+
+        if self._is_idempotent(tool_name):
+            record = self._no_progress.get(signature)
+            if record is not None:
+                _result_hash, repeat_count = record
+                if repeat_count >= self.config.no_progress_block_after:
+                    decision = ToolGuardrailDecision(
+                        action="block",
+                        code="idempotent_no_progress_block",
+                        message=(
+                            f"Blocked {tool_name}: this read-only call returned the same "
+                            f"result {repeat_count} times. Stop repeating it unchanged; "
+                            "use the result already provided or try a different query."
+                        ),
+                        tool_name=tool_name,
+                        count=repeat_count,
+                        signature=signature,
+                    )
+                    self._halt_decision = decision
+                    return decision
+
+        return ToolGuardrailDecision(tool_name=tool_name, signature=signature)
+
+    def after_call(
+        self,
+        tool_name: str,
+        args: Mapping[str, Any] | None,
+        result: str | None,
+        *,
+        failed: bool | None = None,
+    ) -> ToolGuardrailDecision:
+        args = _coerce_args(args)
+        signature = ToolCallSignature.from_call(tool_name, args)
+        if failed is None:
+            failed, _ = classify_tool_failure(tool_name, result)
+
+        if failed:
+            exact_count = self._exact_failure_counts.get(signature, 0) + 1
+            self._exact_failure_counts[signature] = exact_count
+            self._no_progress.pop(signature, None)
+
+            same_count = self._same_tool_failure_counts.get(tool_name, 0) + 1
+            self._same_tool_failure_counts[tool_name] = same_count
+
+            if self.config.hard_stop_enabled and same_count >= self.config.same_tool_failure_halt_after:
+                decision = ToolGuardrailDecision(
+                    action="halt",
+                    code="same_tool_failure_halt",
+                    message=(
+                        f"Stopped {tool_name}: it failed {same_count} times this turn. "
+                        "Stop retrying the same failing tool path and choose a different approach."
+                    ),
+                    tool_name=tool_name,
+                    count=same_count,
+                    signature=signature,
+                )
+                self._halt_decision = decision
+                return decision
+
+            if self.config.warnings_enabled and exact_count >= self.config.exact_failure_warn_after:
+                return ToolGuardrailDecision(
+                    action="warn",
+                    code="repeated_exact_failure_warning",
+                    message=(
+                        f"{tool_name} has failed {exact_count} times with identical arguments. "
+                        "This looks like a loop; inspect the error and change strategy "
+                        "instead of retrying it unchanged."
+                    ),
+                    tool_name=tool_name,
+                    count=exact_count,
+                    signature=signature,
+                )
+
+            if self.config.warnings_enabled and same_count >= self.config.same_tool_failure_warn_after:
+                return ToolGuardrailDecision(
+                    action="warn",
+                    code="same_tool_failure_warning",
+                    message=(
+                        f"{tool_name} has failed {same_count} times this turn. "
+                        "This looks like a loop; change approach before retrying."
+                    ),
+                    tool_name=tool_name,
+                    count=same_count,
+                    signature=signature,
+                )
+
+            return ToolGuardrailDecision(tool_name=tool_name, count=exact_count, signature=signature)
+
+        self._exact_failure_counts.pop(signature, None)
+        self._same_tool_failure_counts.pop(tool_name, None)
+
+        if not self._is_idempotent(tool_name):
+            self._no_progress.pop(signature, None)
+            return ToolGuardrailDecision(tool_name=tool_name, signature=signature)
+
+        result_hash = _result_hash(result)
+        previous = self._no_progress.get(signature)
+        repeat_count = 1
+        if previous is not None and previous[0] == result_hash:
+            repeat_count = previous[1] + 1
+        self._no_progress[signature] = (result_hash, repeat_count)
+
+        if self.config.warnings_enabled and repeat_count >= self.config.no_progress_warn_after:
+            return ToolGuardrailDecision(
+                action="warn",
+                code="idempotent_no_progress_warning",
+                message=(
+                    f"{tool_name} returned the same result {repeat_count} times. "
+                    "Use the result already provided or change the query instead of "
+                    "repeating it unchanged."
+                ),
+                tool_name=tool_name,
+                count=repeat_count,
+                signature=signature,
+            )
+
+        return ToolGuardrailDecision(tool_name=tool_name, count=repeat_count, signature=signature)
+
+    def _is_idempotent(self, tool_name: str) -> bool:
+        if tool_name in self.config.mutating_tools:
+            return False
+        return tool_name in self.config.idempotent_tools
+
+
+def toolguard_synthetic_result(decision: ToolGuardrailDecision) -> str:
+    """Build a synthetic role=tool content string for a blocked tool call."""
+    return json.dumps(
+        {
+            "error": decision.message,
+            "guardrail": decision.to_metadata(),
+        },
+        ensure_ascii=False,
+    )
+
+
+def append_toolguard_guidance(result: str, decision: ToolGuardrailDecision) -> str:
+    """Append runtime guidance to the current tool result content."""
+    if decision.action not in {"warn", "halt"} or not decision.message:
+        return result
+    label = "Tool loop hard stop" if decision.action == "halt" else "Tool loop warning"
+    suffix = (
+        f"\n\n[{label}: "
+        f"{decision.code}; count={decision.count}; {decision.message}]"
+    )
+    return (result or "") + suffix
+
+
+def _coerce_args(args: Mapping[str, Any] | None) -> Mapping[str, Any]:
+    return args if isinstance(args, Mapping) else {}
+
+
+def _result_hash(result: str | None) -> str:
+    parsed = safe_json_loads(result or "")
+    if parsed is not None:
+        try:
+            canonical = json.dumps(
+                parsed,
+                ensure_ascii=False,
+                sort_keys=True,
+                separators=(",", ":"),
+                default=str,
+            )
+        except TypeError:
+            canonical = str(parsed)
+    else:
+        canonical = result or ""
+    return _sha256(canonical)
+
+
+def _as_bool(value: Any, default: bool) -> bool:
+    if value is None:
+        return default
+    if isinstance(value, bool):
+        return value
+    if isinstance(value, (int, float)):
+        return bool(value)
+    if isinstance(value, str):
+        lowered = value.strip().lower()
+        if lowered in {"1", "true", "yes", "on", "enabled"}:
+            return True
+        if lowered in {"0", "false", "no", "off", "disabled"}:
+            return False
+    return default
+
+
+def _positive_int(value: Any, default: int) -> int:
+    if value is None:
+        return default
+    try:
+        parsed = int(value)
+    except (TypeError, ValueError):
+        return default
+    return parsed if parsed >= 1 else default
+
+
+def _sha256(value: str) -> str:
+    return hashlib.sha256(value.encode("utf-8")).hexdigest()

agent/transports/codex.py

@@ -143,7 +143,18 @@ class ResponsesApiTransport(ProviderTransport):
             kwargs["max_output_tokens"] = max_tokens
 
         if is_xai_responses and session_id:
-            kwargs["extra_headers"] = {"x-grok-conv-id": session_id}
+            existing_extra_headers = kwargs.get("extra_headers")
+            merged_extra_headers: Dict[str, str] = {}
+            if isinstance(existing_extra_headers, dict):
+                merged_extra_headers.update(
+                    {
+                        str(key): str(value)
+                        for key, value in existing_extra_headers.items()
+                        if key and value is not None
+                    }
+                )
+            merged_extra_headers["x-grok-conv-id"] = session_id
+            kwargs["extra_headers"] = merged_extra_headers
 
         return kwargs
 

apps/dashboard/README.md

@@ -10,17 +10,34 @@ Browser-based dashboard for managing Hermes Agent configuration, API keys, and m
 
 ## Development
 
-```bash
-# Start the backend API server
-cd ../
-python -m hermes_cli.main web --no-open
+Install workspace dependencies from the repo root first:
 
-# In another terminal, start the Vite dev server (with HMR + API proxy)
-cd web/
+```bash
+npm install
+```
+
+Start the backend API server from the repo root:
+
+```bash
+hermes dashboard --tui --no-open
+```
+
+`--tui` exposes the in-browser Chat tab through `/api/pty`. Omit it if you only need the config/session dashboard.
+
+In another terminal, start the Vite dev server:
+
+```bash
+cd apps/dashboard
 npm run dev
 ```
 
-The Vite dev server proxies `/api` requests to `http://127.0.0.1:9119` (the FastAPI backend).
+The Vite dev server proxies `/api`, `/api/pty`, and `/dashboard-plugins` to `http://127.0.0.1:9119` (the FastAPI backend). It also fetches the backend's `index.html` on each dev page load so the ephemeral session token stays in sync.
+
+If the `hermes` entry point is not installed, use:
+
+```bash
+python -m hermes_cli.main dashboard --tui --no-open
+```
 
 ## Build
 
@@ -28,7 +45,7 @@ The Vite dev server proxies `/api` requests to `http://127.0.0.1:9119` (the Fast
 npm run build
 ```
 
-This outputs to `../hermes_cli/web_dist/`, which the FastAPI server serves as a static SPA. The built assets are included in the Python package via `pyproject.toml` package-data.
+This outputs to `../../hermes_cli/web_dist/`, which the FastAPI server serves as a static SPA. The built assets are included in the Python package via `pyproject.toml` package-data.
 
 ## Structure
 

apps/dashboard/package-lock.json

@@ -1,13 +1,14 @@
 {
-  "name": "web",
+  "name": "dashboard",
   "version": "0.0.0",
   "lockfileVersion": 3,
   "requires": true,
   "packages": {
     "": {
-      "name": "web",
+      "name": "dashboard",
       "version": "0.0.0",
       "dependencies": {
+        "@hermes/shared": "file:../shared",
         "@nous-research/ui": "^0.10.0",
         "@observablehq/plot": "^0.6.17",
         "@react-three/fiber": "^9.6.0",
@@ -45,6 +46,13 @@
         "vite": "^7.3.1"
       }
     },
+    "../shared": {
+      "name": "@hermes/shared",
+      "version": "0.0.0",
+      "devDependencies": {
+        "typescript": "^6.0.3"
+      }
+    },
     "node_modules/@babel/code-frame": {
       "version": "7.29.0",
       "resolved": "https://registry.npmjs.org/@babel/code-frame/-/code-frame-7.29.0.tgz",
@@ -947,6 +955,10 @@
       "integrity": "sha512-RiB/yIh78pcIxl6lLMG0CgBXAZ2Y0eVHqMPYugu+9U0AeT6YBeiJpf7lbdJNIugFP5SIjwNRgo4DhR1Qxi26Gg==",
       "license": "MIT"
     },
+    "node_modules/@hermes/shared": {
+      "resolved": "../shared",
+      "link": true
+    },
     "node_modules/@humanfs/core": {
       "version": "0.19.2",
       "resolved": "https://registry.npmjs.org/@humanfs/core/-/core-0.19.2.tgz",
@@ -2371,6 +2383,64 @@
         "node": ">=14.0.0"
       }
     },
+    "node_modules/@tailwindcss/oxide-wasm32-wasi/node_modules/@emnapi/core": {
+      "version": "1.8.1",
+      "inBundle": true,
+      "license": "MIT",
+      "optional": true,
+      "dependencies": {
+        "@emnapi/wasi-threads": "1.1.0",
+        "tslib": "^2.4.0"
+      }
+    },
+    "node_modules/@tailwindcss/oxide-wasm32-wasi/node_modules/@emnapi/runtime": {
+      "version": "1.8.1",
+      "inBundle": true,
+      "license": "MIT",
+      "optional": true,
+      "dependencies": {
+        "tslib": "^2.4.0"
+      }
+    },
+    "node_modules/@tailwindcss/oxide-wasm32-wasi/node_modules/@emnapi/wasi-threads": {
+      "version": "1.1.0",
+      "inBundle": true,
+      "license": "MIT",
+      "optional": true,
+      "dependencies": {
+        "tslib": "^2.4.0"
+      }
+    },
+    "node_modules/@tailwindcss/oxide-wasm32-wasi/node_modules/@napi-rs/wasm-runtime": {
+      "version": "1.1.1",
+      "inBundle": true,
+      "license": "MIT",
+      "optional": true,
+      "dependencies": {
+        "@emnapi/core": "^1.7.1",
+        "@emnapi/runtime": "^1.7.1",
+        "@tybys/wasm-util": "^0.10.1"
+      },
+      "funding": {
+        "type": "github",
+        "url": "https://github.com/sponsors/Brooooooklyn"
+      }
+    },
+    "node_modules/@tailwindcss/oxide-wasm32-wasi/node_modules/@tybys/wasm-util": {
+      "version": "0.10.1",
+      "inBundle": true,
+      "license": "MIT",
+      "optional": true,
+      "dependencies": {
+        "tslib": "^2.4.0"
+      }
+    },
+    "node_modules/@tailwindcss/oxide-wasm32-wasi/node_modules/tslib": {
+      "version": "2.8.1",
+      "inBundle": true,
+      "license": "0BSD",
+      "optional": true
+    },
     "node_modules/@tailwindcss/oxide-win32-arm64-msvc": {
       "version": "4.2.4",
       "resolved": "https://registry.npmjs.org/@tailwindcss/oxide-win32-arm64-msvc/-/oxide-win32-arm64-msvc-4.2.4.tgz",

apps/dashboard/package.json

@@ -1,10 +1,10 @@
 {
-  "name": "web",
+  "name": "dashboard",
   "private": true,
   "version": "0.0.0",
   "type": "module",
   "scripts": {
-    "sync-assets": "rm -rf public/fonts public/ds-assets && cp -r node_modules/@nous-research/ui/dist/fonts public/fonts && cp -r node_modules/@nous-research/ui/dist/assets public/ds-assets",
+    "sync-assets": "node scripts/sync-assets.cjs",
     "predev": "npm run sync-assets",
     "prebuild": "npm run sync-assets",
     "dev": "vite",
@@ -13,6 +13,7 @@
     "preview": "vite preview"
   },
   "dependencies": {
+    "@hermes/shared": "file:../shared",
     "@nous-research/ui": "^0.10.0",
     "@observablehq/plot": "^0.6.17",
     "@react-three/fiber": "^9.6.0",

apps/dashboard/scripts/sync-assets.cjs

@@ -0,0 +1,46 @@
+#!/usr/bin/env node
+/**
+ * Copy font and asset folders from @nous-research/ui into public/ for Vite.
+ *
+ * Locates @nous-research/ui by walking up from this script looking for
+ * node_modules/@nous-research/ui — works whether the dep is co-located
+ * (non-workspace layout) or hoisted to the repo root (npm workspaces).
+ */
+const fs = require('node:fs')
+const path = require('node:path')
+
+const DASHBOARD_ROOT = path.resolve(__dirname, '..')
+
+function locateUiPackage() {
+  let dir = DASHBOARD_ROOT
+  const { root } = path.parse(dir)
+  while (true) {
+    const candidate = path.join(dir, 'node_modules', '@nous-research', 'ui')
+    if (fs.existsSync(path.join(candidate, 'package.json'))) {
+      return candidate
+    }
+    if (dir === root) break
+    dir = path.dirname(dir)
+  }
+  throw new Error(
+    '@nous-research/ui not found. Run `npm install` from the repo root.'
+  )
+}
+
+const uiRoot = locateUiPackage()
+const distRoot = path.join(uiRoot, 'dist')
+
+const mappings = [
+  ['fonts', path.join(DASHBOARD_ROOT, 'public', 'fonts')],
+  ['assets', path.join(DASHBOARD_ROOT, 'public', 'ds-assets')],
+]
+
+for (const [srcName, destPath] of mappings) {
+  const srcPath = path.join(distRoot, srcName)
+  if (!fs.existsSync(srcPath)) {
+    throw new Error(`Missing ${srcPath} in @nous-research/ui — rebuild that package.`)
+  }
+  fs.rmSync(destPath, { recursive: true, force: true })
+  fs.cpSync(srcPath, destPath, { recursive: true })
+  console.log(`synced ${path.relative(DASHBOARD_ROOT, destPath)}`)
+}

apps/dashboard/src/App.tsx

@@ -65,10 +65,12 @@ import ModelsPage from "@/pages/ModelsPage";
 import CronPage from "@/pages/CronPage";
 import ProfilesPage from "@/pages/ProfilesPage";
 import SkillsPage from "@/pages/SkillsPage";
+import PluginsPage from "@/pages/PluginsPage";
 import ChatPage from "@/pages/ChatPage";
 import { LanguageSwitcher } from "@/components/LanguageSwitcher";
 import { ThemeSwitcher } from "@/components/ThemeSwitcher";
 import { useI18n } from "@/i18n";
+import type { Translations } from "@/i18n/types";
 import { PluginPage, PluginSlot, usePlugins } from "@/plugins";
 import type { PluginManifest } from "@/plugins";
 import { useTheme } from "@/themes";
@@ -78,6 +80,14 @@ function RootRedirect() {
   return <Navigate to="/sessions" replace />;
 }
 
+function UnknownRouteFallback({ pluginsLoading }: { pluginsLoading: boolean }) {
+  if (pluginsLoading) {
+    // Render nothing during the plugin-load window — a spinner here would just flash.
+    return null;
+  }
+  return <Navigate to="/sessions" replace />;
+}
+
 const CHAT_NAV_ITEM: NavItem = {
   path: "/chat",
   labelKey: "chat",
@@ -102,6 +112,7 @@ const BUILTIN_ROUTES_CORE: Record<string, ComponentType> = {
   "/logs": LogsPage,
   "/cron": CronPage,
   "/skills": SkillsPage,
+  "/plugins": PluginsPage,
   "/profiles": ProfilesPage,
   "/config": ConfigPage,
   "/env": EnvPage,
@@ -138,6 +149,7 @@ const BUILTIN_NAV_REST: NavItem[] = [
   { path: "/logs", labelKey: "logs", label: "Logs", icon: FileText },
   { path: "/cron", labelKey: "cron", label: "Cron", icon: Clock },
   { path: "/skills", labelKey: "skills", label: "Skills", icon: Package },
+  { path: "/plugins", labelKey: "plugins", label: "Plugins", icon: Puzzle },
   { path: "/profiles", labelKey: "profiles", label: "Profiles", icon: Users },
   { path: "/config", labelKey: "config", label: "Config", icon: Settings },
   { path: "/env", labelKey: "keys", label: "Keys", icon: KeyRound },
@@ -213,6 +225,22 @@ function buildNavItems(
   return items;
 }
 
+/** Split merged nav into built-in sidebar entries vs plugin tabs, preserving plugin order hints. */
+function partitionSidebarNav(
+  builtIn: NavItem[],
+  manifests: PluginManifest[],
+): { coreItems: NavItem[]; pluginItems: NavItem[] } {
+  const merged = buildNavItems(builtIn, manifests);
+  const builtinPaths = new Set(builtIn.map((i) => i.path));
+  const coreItems: NavItem[] = [];
+  const pluginItems: NavItem[] = [];
+  for (const item of merged) {
+    if (builtinPaths.has(item.path)) coreItems.push(item);
+    else pluginItems.push(item);
+  }
+  return { coreItems, pluginItems };
+}
+
 function buildRoutes(
   builtinRoutes: Record<string, ComponentType>,
   manifests: PluginManifest[],
@@ -253,6 +281,7 @@ function buildRoutes(
 
   for (const m of addons) {
     if (m.tab.hidden) continue;
+    if (m.tab.path === "/plugins") continue;
     if (builtinRoutes[m.tab.path]) continue;
     routes.push({
       key: `plugin:${m.name}`,
@@ -263,6 +292,7 @@ function buildRoutes(
 
   for (const m of manifests) {
     if (!m.tab.hidden) continue;
+    if (m.tab.path === "/plugins") continue;
     if (builtinRoutes[m.tab.path] || m.tab.override) continue;
     routes.push({
       key: `plugin:hidden:${m.name}`,
@@ -322,8 +352,8 @@ export default function App() {
     [embeddedChat],
   );
 
-  const navItems = useMemo(
-    () => buildNavItems(builtinNav, manifests),
+  const sidebarNav = useMemo(
+    () => partitionSidebarNav(builtinNav, manifests),
     [builtinNav, manifests],
   );
   const routes = useMemo(
@@ -476,56 +506,44 @@ export default function App() {
               aria-label={t.app.navigation}
             >
               <ul className="flex flex-col">
-                {navItems.map(({ path, label, labelKey, icon: Icon }) => {
-                  const navLabel = labelKey
-                    ? ((t.app.nav as Record<string, string>)[labelKey] ?? label)
-                    : label;
-                  return (
-                    <li key={path}>
-                      <NavLink
-                        to={path}
-                        end={path === "/sessions"}
-                        onClick={closeMobile}
-                        className={({ isActive }) =>
-                          cn(
-                            "group relative flex items-center gap-3",
-                            "px-5 py-2.5",
-                            "font-mondwest text-[0.8rem] tracking-[0.12em]",
-                            "whitespace-nowrap transition-colors cursor-pointer",
-                            "focus-visible:outline-none focus-visible:ring-1 focus-visible:ring-midground",
-                            isActive
-                              ? "text-midground"
-                              : "opacity-60 hover:opacity-100",
-                          )
-                        }
-                        style={{
-                          clipPath: "var(--component-tab-clip-path)",
-                        }}
-                      >
-                        {({ isActive }) => (
-                          <>
-                            <Icon className="h-3.5 w-3.5 shrink-0" />
-                            <span className="truncate">{navLabel}</span>
-
-                            <span
-                              aria-hidden
-                              className="absolute inset-y-0.5 left-1.5 right-1.5 bg-midground opacity-0 pointer-events-none transition-opacity duration-200 group-hover:opacity-5"
-                            />
-
-                            {isActive && (
-                              <span
-                                aria-hidden
-                                className="absolute left-0 top-0 bottom-0 w-px bg-midground"
-                                style={{ mixBlendMode: "plus-lighter" }}
-                              />
-                            )}
-                          </>
-                        )}
-                      </NavLink>
-                    </li>
-                  );
-                })}
+                {sidebarNav.coreItems.map((item) => (
+                  <SidebarNavLink
+                    closeMobile={closeMobile}
+                    item={item}
+                    key={item.path}
+                    t={t}
+                  />
+                ))}
               </ul>
+
+              {sidebarNav.pluginItems.length > 0 && (
+                <div
+                  aria-labelledby="hermes-sidebar-plugin-nav-heading"
+                  className="flex flex-col border-t border-current/10 pb-2"
+                  role="group"
+                >
+                  <span
+                    className={cn(
+                      "px-5 pt-2.5 pb-1",
+                      "font-mondwest text-[0.6rem] tracking-[0.15em] uppercase opacity-30",
+                    )}
+                    id="hermes-sidebar-plugin-nav-heading"
+                  >
+                    {t.app.pluginNavSection}
+                  </span>
+
+                  <ul className="flex flex-col">
+                    {sidebarNav.pluginItems.map((item) => (
+                      <SidebarNavLink
+                        closeMobile={closeMobile}
+                        item={item}
+                        key={item.path}
+                        t={t}
+                      />
+                    ))}
+                  </ul>
+                </div>
+              )}
             </nav>
 
             <SidebarSystemActions onNavigate={closeMobile} />
@@ -572,7 +590,9 @@ export default function App() {
                   ))}
                   <Route
                     path="*"
-                    element={<Navigate to="/sessions" replace />}
+                    element={
+                      <UnknownRouteFallback pluginsLoading={pluginsLoading} />
+                    }
                   />
                 </Routes>
 
@@ -615,6 +635,57 @@ export default function App() {
   );
 }
 
+function SidebarNavLink({ closeMobile, item, t }: SidebarNavLinkProps) {
+  const { path, label, labelKey, icon: Icon } = item;
+
+  const navLabel = labelKey
+    ? ((t.app.nav as Record<string, string>)[labelKey] ?? label)
+    : label;
+
+  return (
+    <li>
+      <NavLink
+        to={path}
+        end={path === "/sessions"}
+        onClick={closeMobile}
+        className={({ isActive }) =>
+          cn(
+            "group relative flex items-center gap-3",
+            "px-5 py-2.5",
+            "font-mondwest text-[0.8rem] tracking-[0.12em]",
+            "whitespace-nowrap transition-colors cursor-pointer",
+            "focus-visible:outline-none focus-visible:ring-1 focus-visible:ring-midground",
+            isActive ? "text-midground" : "opacity-60 hover:opacity-100",
+          )
+        }
+        style={{
+          clipPath: "var(--component-tab-clip-path)",
+        }}
+      >
+        {({ isActive }) => (
+          <>
+            <Icon className="h-3.5 w-3.5 shrink-0" />
+            <span className="truncate">{navLabel}</span>
+
+            <span
+              aria-hidden
+              className="absolute inset-y-0.5 left-1.5 right-1.5 bg-midground opacity-0 pointer-events-none transition-opacity duration-200 group-hover:opacity-5"
+            />
+
+            {isActive && (
+              <span
+                aria-hidden
+                className="absolute left-0 top-0 bottom-0 w-px bg-midground"
+                style={{ mixBlendMode: "plus-lighter" }}
+              />
+            )}
+          </>
+        )}
+      </NavLink>
+    </li>
+  );
+}
+
 function SidebarSystemActions({ onNavigate }: { onNavigate: () => void }) {
   const { t } = useI18n();
   const navigate = useNavigate();
@@ -733,6 +804,12 @@ interface NavItem {
   path: string;
 }
 
+interface SidebarNavLinkProps {
+  closeMobile: () => void;
+  item: NavItem;
+  t: Translations;
+}
+
 interface SystemActionItem {
   action: SystemAction;
   icon: ComponentType<{ className?: string }>;

apps/dashboard/src/components/ThemeSwitcher.tsx

@@ -4,6 +4,7 @@ import { Button } from "@nous-research/ui/ui/components/button";
 import { ListItem } from "@nous-research/ui/ui/components/list-item";
 import { Typography } from "@/components/NouiTypography";
 import { BUILTIN_THEMES, useTheme } from "@/themes";
+import type { DashboardTheme } from "@/themes";
 import { useI18n } from "@/i18n";
 import { cn } from "@/lib/utils";
 
@@ -11,8 +12,8 @@ import { cn } from "@/lib/utils";
  * Compact theme picker mounted next to the language switcher in the header.
  * Each dropdown row shows a 3-stop swatch (background / midground / warm
  * glow) so users can preview the palette before committing. User-defined
- * themes from `~/.hermes/dashboard-themes/*.yaml` that aren't in
- * `BUILTIN_THEMES` render without swatches and apply the default palette.
+ * themes from `~/.hermes/dashboard-themes/*.yaml` use their API-provided
+ * definitions so they show real palette swatches just like built-ins.
  *
  * When placed at the bottom of a container (e.g. the sidebar rail), pass
  * `dropUp` so the menu opens above the trigger instead of clipping below
@@ -95,7 +96,7 @@ export function ThemeSwitcher({ dropUp = false }: ThemeSwitcherProps) {
 
           {availableThemes.map((th) => {
             const isActive = th.name === themeName;
-            const preset = BUILTIN_THEMES[th.name];
+            const paletteTheme = BUILTIN_THEMES[th.name] ?? th.definition;
 
             return (
               <ListItem
@@ -109,8 +110,8 @@ export function ThemeSwitcher({ dropUp = false }: ThemeSwitcherProps) {
                 }}
                 className="gap-3"
               >
-                {preset ? (
-                  <ThemeSwatch theme={preset.name} />
+                {paletteTheme ? (
+                  <ThemeSwatch theme={paletteTheme} />
                 ) : (
                   <PlaceholderSwatch />
                 )}
@@ -144,10 +145,8 @@ export function ThemeSwitcher({ dropUp = false }: ThemeSwitcherProps) {
   );
 }
 
-function ThemeSwatch({ theme }: { theme: string }) {
-  const preset = BUILTIN_THEMES[theme];
-  if (!preset) return <PlaceholderSwatch />;
-  const { background, midground, warmGlow } = preset.palette;
+function ThemeSwatch({ theme }: { theme: DashboardTheme }) {
+  const { background, midground, warmGlow } = theme.palette;
   return (
     <div
       aria-hidden

apps/dashboard/src/i18n/en.ts

@@ -76,6 +76,7 @@ export const en: Translations = {
       logs: "Logs",
       models: "Models",
       profiles: "profiles : multi agents",
+      plugins: "Plugins",
       sessions: "Sessions",
       skills: "Skills",
     },
@@ -84,6 +85,7 @@ export const en: Translations = {
     navigation: "Navigation",
     openDocumentation: "Open documentation in a new tab",
     openNavigation: "Open navigation",
+    pluginNavSection: "Plugins",
     sessionsActiveCount: "{count} active",
     statusOverview: "Status overview",
     system: "System",
@@ -256,6 +258,47 @@ export const en: Translations = {
     renamed: "Renamed",
   },
 
+  pluginsPage: {
+    contextEngineLabel: "Context engine",
+    dashboardSlots: "Dashboard slots",
+    disableRuntime: "Disable",
+    enableAfterInstall: "Enable after install",
+    enableRuntime: "Enable",
+    forceReinstall: "Force reinstall (delete existing folder first)",
+    headline:
+      "Discover, install, enable, and update Hermes plugins (`hermes plugins` parity).",
+    identifierLabel: "Git URL or owner/repo",
+    inactive: "inactive",
+    installBtn: "Install from Git",
+    installHeading: "Install from GitHub / Git URL",
+    installHint: "Use owner/repo shorthand or a full https:// or git@ clone URL.",
+    memoryProviderLabel: "Memory provider",
+    missingEnvWarn: "Set these in Keys before the plugin can run:",
+    noDashboardTab: "No dashboard tab",
+    openTab: "Open",
+    orphanHeading: "Dashboard-only extensions (no agent plugin.yaml match)",
+    pluginListHeading: "Installed plugins",
+    providerDefaults: "built-in / default",
+    providersHeading: "Runtime provider plugins",
+    providersHint:
+      "Writes memory.provider (empty = built-in) and context.engine to config.yaml. Takes effect next session.",
+    refreshDashboard: "Rescan dashboard extensions",
+    removeConfirm: "Remove this plugin from ~/.hermes/plugins/?",
+    removeHint: "Only user-installed plugins under ~/.hermes/plugins can be removed.",
+    rescanHeading: "SPA plugin registry",
+    rescanHint: "Rescan after adding files on disk so the dashboard sidebar picks up new manifests.",
+    runtimeHeading: "Gateway runtime (YAML plugins)",
+    saveProviders: "Save provider settings",
+    savedProviders: "Provider settings saved.",
+    sourceBadge: "Source",
+    authRequired: "Auth required",
+    authRequiredHint: "Run this command to authenticate:",
+    updateGit: "Git pull",
+    versionBadge: "Version",
+    showInSidebar: "Show in sidebar",
+    hideFromSidebar: "Hide from sidebar",
+  },
+
   skills: {
     title: "Skills",
     searchPlaceholder: "Search skills and toolsets...",

apps/dashboard/src/i18n/types.ts

@@ -76,6 +76,7 @@ export interface Translations {
       logs: string;
       models: string;
       profiles: string;
+      plugins: string;
       sessions: string;
       skills: string;
     };
@@ -84,6 +85,7 @@ export interface Translations {
     navigation: string;
     openDocumentation: string;
     openNavigation: string;
+    pluginNavSection: string;
     sessionsActiveCount: string;
     statusOverview: string;
     system: string;
@@ -228,6 +230,46 @@ export interface Translations {
     };
   };
 
+  // ── Plugins page ──
+  pluginsPage: {
+    contextEngineLabel: string;
+    dashboardSlots: string;
+    disableRuntime: string;
+    enableAfterInstall: string;
+    enableRuntime: string;
+    forceReinstall: string;
+    headline: string;
+    identifierLabel: string;
+    inactive: string;
+    installBtn: string;
+    installHeading: string;
+    installHint: string;
+    memoryProviderLabel: string;
+    missingEnvWarn: string;
+    noDashboardTab: string;
+    openTab: string;
+    orphanHeading: string;
+    pluginListHeading: string;
+    providerDefaults: string;
+    providersHeading: string;
+    providersHint: string;
+    refreshDashboard: string;
+    removeConfirm: string;
+    removeHint: string;
+    rescanHeading: string;
+    rescanHint: string;
+    runtimeHeading: string;
+    saveProviders: string;
+    savedProviders: string;
+    sourceBadge: string;
+    authRequired: string;
+    authRequiredHint: string;
+    updateGit: string;
+    versionBadge: string;
+    showInSidebar: string;
+    hideFromSidebar: string;
+  };
+
   // ── Profiles page ──
   profiles: {
     newProfile: string;

apps/dashboard/src/i18n/zh.ts

@@ -75,6 +75,7 @@ export const zh: Translations = {
       logs: "日志",
       models: "模型",
       profiles: "多Agent配置",
+      plugins: "插件管理",
       sessions: "会话",
       skills: "技能",
     },
@@ -83,6 +84,7 @@ export const zh: Translations = {
     navigation: "导航",
     openDocumentation: "在新标签页中打开文档",
     openNavigation: "打开导航",
+    pluginNavSection: "插件",
     sessionsActiveCount: "{count} 个活跃",
     statusOverview: "状态概览",
     system: "系统",
@@ -253,6 +255,46 @@ export const zh: Translations = {
     renamed: "已重命名",
   },
 
+  pluginsPage: {
+    contextEngineLabel: "上下文引擎",
+    dashboardSlots: "面板插槽",
+    disableRuntime: "禁用",
+    enableAfterInstall: "安装后启用",
+    enableRuntime: "启用",
+    forceReinstall: "强制重装（先删除已有目录）",
+    headline: "发现、安装、启用和更新 Hermes 插件（对齐 `hermes plugins` CLI）。",
+    identifierLabel: "Git 地址或 owner/repo",
+    inactive: "未启用",
+    installBtn: "从 Git 安装",
+    installHeading: "从 GitHub / Git 地址安装",
+    installHint: "使用 owner/repo 简写或完整的 https:// / git@ 克隆地址。",
+    memoryProviderLabel: "记忆提供方",
+    missingEnvWarn: "在「密钥」页面设置以下变量后再运行插件：",
+    noDashboardTab: "无仪表盘标签",
+    openTab: "打开",
+    orphanHeading: "仅仪表盘扩展（无匹配的 agent plugin.yaml）",
+    pluginListHeading: "已安装插件",
+    providerDefaults: "内置 / 默认",
+    providersHeading: "运行时提供方插件",
+    providersHint:
+      "写入 config.yaml：memory.provider（留空为内置）、context.engine。下次会话生效。",
+    refreshDashboard: "重新扫描仪表盘扩展",
+    removeConfirm: "从 ~/.hermes/plugins/ 删除此插件？",
+    removeHint: "仅可移除用户安装在 ~/.hermes/plugins 下的插件。",
+    rescanHeading: "SPA 插件注册表",
+    rescanHint: "在磁盘新增文件后扫描，使侧边栏载入新 manifest。",
+    runtimeHeading: "网关运行时（YAML 插件）",
+    saveProviders: "保存提供方设置",
+    savedProviders: "提供方设置已保存。",
+    sourceBadge: "来源",
+    authRequired: "需要认证",
+    authRequiredHint: "运行此命令以完成认证：",
+    updateGit: "git pull",
+    versionBadge: "版本",
+    showInSidebar: "在侧边栏显示",
+    hideFromSidebar: "从侧边栏隐藏",
+  },
+
   skills: {
     title: "技能",
     searchPlaceholder: "搜索技能和工具集...",

apps/dashboard/src/lib/api.ts

@@ -259,6 +259,56 @@ export const api = {
   rescanPlugins: () =>
     fetchJSON<{ ok: boolean; count: number }>("/api/dashboard/plugins/rescan"),
 
+  getPluginsHub: () => fetchJSON<PluginsHubResponse>("/api/dashboard/plugins/hub"),
+
+  installAgentPlugin: (body: AgentPluginInstallRequest) =>
+    fetchJSON<AgentPluginInstallResponse>("/api/dashboard/agent-plugins/install", {
+      method: "POST",
+      headers: { "Content-Type": "application/json" },
+      body: JSON.stringify({ ...body }),
+    }),
+
+  enableAgentPlugin: (name: string) =>
+    fetchJSON<{ ok: boolean; name: string; unchanged?: boolean }>(
+      `/api/dashboard/agent-plugins/${encodeURIComponent(name)}/enable`,
+      { method: "POST" },
+    ),
+
+  disableAgentPlugin: (name: string) =>
+    fetchJSON<{ ok: boolean; name: string; unchanged?: boolean }>(
+      `/api/dashboard/agent-plugins/${encodeURIComponent(name)}/disable`,
+      { method: "POST" },
+    ),
+
+  updateAgentPlugin: (name: string) =>
+    fetchJSON<AgentPluginUpdateResponse>(
+      `/api/dashboard/agent-plugins/${encodeURIComponent(name)}/update`,
+      { method: "POST" },
+    ),
+
+  removeAgentPlugin: (name: string) =>
+    fetchJSON<{ ok: boolean; name: string }>(
+      `/api/dashboard/agent-plugins/${encodeURIComponent(name)}`,
+      { method: "DELETE" },
+    ),
+
+  savePluginProviders: (body: PluginProvidersPutRequest) =>
+    fetchJSON<{ ok: boolean }>("/api/dashboard/plugin-providers", {
+      method: "PUT",
+      headers: { "Content-Type": "application/json" },
+      body: JSON.stringify(body),
+    }),
+
+  setPluginVisibility: (name: string, hidden: boolean) =>
+    fetchJSON<{ ok: boolean; name: string; hidden: boolean }>(
+      `/api/dashboard/plugins/${encodeURIComponent(name)}/visibility`,
+      {
+        method: "POST",
+        headers: { "Content-Type": "application/json" },
+        body: JSON.stringify({ hidden }),
+      },
+    ),
+
   // Dashboard themes
   getThemes: () =>
     fetchJSON<DashboardThemesResponse>("/api/dashboard/themes"),
@@ -668,8 +718,67 @@ export interface PluginManifestResponse {
     override?: string;
     hidden?: boolean;
   };
+  slots?: string[];
   entry: string;
   css?: string | null;
   has_api: boolean;
   source: string;
 }
+
+export interface HubAgentPluginRow {
+  name: string;
+  version: string;
+  description: string;
+  source: string;
+  runtime_status: "disabled" | "enabled" | "inactive";
+  has_dashboard_manifest: boolean;
+  dashboard_manifest: PluginManifestResponse | null;
+  path: string;
+  can_remove: boolean;
+  can_update_git: boolean;
+  auth_required: boolean;
+  auth_command: string;
+  user_hidden: boolean;
+}
+
+export interface PluginsHubProviders {
+  memory_provider: string;
+  memory_options: Array<{ name: string; description: string }>;
+  context_engine: string;
+  context_options: Array<{ name: string; description: string }>;
+}
+
+export interface PluginsHubResponse {
+  plugins: HubAgentPluginRow[];
+  orphan_dashboard_plugins: PluginManifestResponse[];
+  providers: PluginsHubProviders;
+}
+
+export interface AgentPluginInstallRequest {
+  identifier: string;
+  force?: boolean;
+  enable?: boolean;
+}
+
+export interface AgentPluginInstallResponse {
+  ok: boolean;
+  plugin_name?: string;
+  warnings?: string[];
+  missing_env?: string[];
+  after_install_path?: string | null;
+  enabled?: boolean;
+  error?: string;
+}
+
+export interface AgentPluginUpdateResponse {
+  ok: boolean;
+  name?: string;
+  output?: string;
+  unchanged?: boolean;
+  error?: string;
+}
+
+export interface PluginProvidersPutRequest {
+  memory_provider?: string;
+  context_engine?: string;
+}

apps/dashboard/src/lib/gatewayClient.ts

@@ -0,0 +1,36 @@
+import {
+  JsonRpcGatewayClient,
+  type ConnectionState,
+  type GatewayEvent,
+  type GatewayEventName,
+} from "@hermes/shared";
+
+export type { ConnectionState, GatewayEvent, GatewayEventName };
+
+/**
+ * Browser wrapper for the shared tui_gateway JSON-RPC client.
+ *
+ * Dashboard resolves its token and host from the served page. Desktop uses the
+ * same shared protocol client, but supplies an absolute wsUrl from Electron.
+ */
+export class GatewayClient extends JsonRpcGatewayClient {
+  async connect(token?: string): Promise<void> {
+    const resolved = token ?? window.__HERMES_SESSION_TOKEN__ ?? "";
+    if (!resolved) {
+      throw new Error(
+        "Session token not available — page must be served by the Hermes dashboard",
+      );
+    }
+
+    const scheme = location.protocol === "https:" ? "wss:" : "ws:";
+    await super.connect(
+      `${scheme}//${location.host}/api/ws?token=${encodeURIComponent(resolved)}`,
+    );
+  }
+}
+
+declare global {
+  interface Window {
+    __HERMES_SESSION_TOKEN__?: string;
+  }
+}