diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index c02a436efb03..595569a82fa0 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -165,3 +165,67 @@ jobs:
sys.exit(1)
print('All checks passed (or were skipped)')
"
+
+ # ─────────────────────────────────────────────────────────────────────
+ # CI timing report: collect per-job/step durations from the GitHub API,
+ # cache them on main (as a baseline), and on PRs generate an HTML diff
+ # report with a gantt chart + per-step breakdown. The report is uploaded
+ # as an artifact and a markdown summary is written to $GITHUB_STEP_SUMMARY.
+ # ─────────────────────────────────────────────────────────────────────
+ ci-timings:
+ name: CI timing report
+ needs: [all-checks-pass, docker]
+ if: always()
+ runs-on: ubuntu-latest
+ steps:
+ - name: Checkout code
+ uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2
+
+ - name: Restore baseline cache (PR only)
+ if: github.event_name == 'pull_request'
+ uses: actions/cache/restore@27d5ce7f107fe9357f9df03efb73ab90386fccae # v5.0.5
+ with:
+ path: ci-timings-baseline.json
+ # Prefix-match: exact key will never hit (run_id differs), so
+ # restore-keys finds the most recent baseline from main.
+ key: ci-timings-baseline-never-exact
+ restore-keys: |
+ ci-timings-baseline-
+
+ - name: Collect timings and generate report
+ env:
+ GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+ run: |
+ python3 scripts/ci/timings_report.py \
+ --baseline ci-timings-baseline.json \
+ --output ci-timings-report.html \
+ --json-out ci-timings.json \
+ --summary-out ci-timings-summary.md
+
+ - name: Upload HTML report
+ uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7
+ id: ci-timings-artifact
+ with:
+ name: ci-timings-report
+ path: ci-timings-report.html
+ retention-days: 14
+ archive: false
+
+ - name: Output summary
+ env:
+ REPORT_URL: ${{ steps.ci-timings-artifact.outputs.artifact-url}}
+ run: |
+ echo "# CI Timing report" >> "$GITHUB_STEP_SUMMARY"
+ echo "[View the full interactive report]($REPORT_URL)" >> "$GITHUB_STEP_SUMMARY"
+ cat ci-timings-summary.md >> "$GITHUB_STEP_SUMMARY"
+
+ - name: Save baseline cache (main only)
+ if: github.event_name == 'push' && github.ref == 'refs/heads/main'
+ run: cp ci-timings.json ci-timings-baseline.json
+
+ - name: Upload baseline to cache (main only)
+ if: github.event_name == 'push' && github.ref == 'refs/heads/main'
+ uses: actions/cache/save@27d5ce7f107fe9357f9df03efb73ab90386fccae # v5.0.5
+ with:
+ path: ci-timings-baseline.json
+ key: ci-timings-baseline-${{ github.run_id }}
diff --git a/.github/workflows/docker.yml b/.github/workflows/docker.yml
index 13b86722b893..8030b889e246 100644
--- a/.github/workflows/docker.yml
+++ b/.github/workflows/docker.yml
@@ -7,15 +7,11 @@ on:
permissions:
contents: read
- # Needed so the arm64 job can push/pull its registry-backed build cache
- # to ghcr.io (cache-to/cache-from type=registry). See the build-arm64
- # job for why registry cache replaced the gha cache on that arch.
- packages: write
# Concurrency: push/release runs are NEVER cancelled so every merge gets
# its own image. PR runs reuse a PR-scoped group with
-# cancel-in-progress: true so rapid pushes to the same PR collapse to the
-# latest commit.
+# cancel-in-progress: true so rapid pushes to the same PR collapse to
+# the latest commit.
concurrency:
group: docker-${{ github.event.pull_request.number || github.ref }}
cancel-in-progress: ${{ github.event_name == 'pull_request' }}
@@ -24,79 +20,47 @@ env:
IMAGE_NAME: nousresearch/hermes-agent
jobs:
- # Build, test, and optionally push the amd64 image.
- build-amd64:
- # Only run on the upstream repository, not on forks
+ # Build, test, and optionally push the image for each architecture.
+ build:
if: github.repository == 'NousResearch/hermes-agent'
- runs-on: ubuntu-latest
+ strategy:
+ fail-fast: false
+ matrix:
+ include:
+ - arch: amd64
+ runner: ubuntu-latest
+ platform: linux/amd64
+ cache-from: type=gha,scope=docker-amd64
+ cache-to: type=gha,mode=max,scope=docker-amd64
+ - arch: arm64
+ runner: ubuntu-24.04-arm
+ platform: linux/arm64
+ cache-from: type=gha,scope=docker-arm64
+ cache-to: type=gha,mode=max,scope=docker-arm64
+
+ runs-on: ${{ matrix.runner }}
timeout-minutes: 45
- outputs:
- digest: ${{ steps.push.outputs.digest }}
steps:
- name: Checkout code
uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2
- # The image build + integration tests run on every event
- # (PRs, push-to-main, release). Publish steps below are gated to
- # push-to-main / release only.
- name: Set up Docker Buildx
uses: docker/setup-buildx-action@8d2750c68a42422c14e847fe6c8ac0403b4cbd6f # v3
# Build once, load into the local daemon for testing. Cached
- # to gha with a per-arch scope; the push step below reuses every
- # layer from this build.
- - name: Build image (amd64)
+ # per-arch; the push step below reuses every layer from this build.
+ - name: Build image (${{ matrix.arch }})
uses: docker/build-push-action@bcafcacb16a39f128d818304e6c9c0c18556b85f # v7.1.0
with:
context: .
file: Dockerfile
load: true
- platforms: linux/amd64
+ platforms: ${{ matrix.platform }}
tags: ${{ env.IMAGE_NAME }}:test
build-args: |
HERMES_GIT_SHA=${{ github.sha }}
- cache-from: type=gha,scope=docker-amd64
- cache-to: type=gha,mode=max,scope=docker-amd64
-
- # Run the docker-integration test suite against the freshly-built
- # image already loaded into the local daemon (`:test`).
- #
- # Piggybacking here avoids a second image build: the build step
- # already loaded the image into the daemon under
- # `${IMAGE_NAME}:test`, so we just point ``HERMES_TEST_IMAGE`` at
- # that. The fixture's ``HERMES_TEST_IMAGE`` branch (see
- # tests/docker/conftest.py:62-63) short-circuits the rebuild.
- #
- # Why this job and not a standalone one: the image is 5GB+; passing
- # it between jobs via ``docker save``/``upload-artifact`` is slower
- # than the build itself. Reusing the existing daemon state is the
- # cheapest path to coverage on every PR that touches docker code.
- # ---------------------------------------------------------------------
- - name: Install uv (for docker tests)
- uses: astral-sh/setup-uv@fac544c07dec837d0ccb6301d7b5580bf5edae39 # 8.2.0
-
- - name: Set up Python 3.11 (for docker tests)
- run: uv python install 3.11
-
- - name: Install Python dependencies (for docker tests)
- run: |
- # ``dev`` extra pulls in pytest, pytest-asyncio —
- # everything tests/docker/ needs. We deliberately avoid ``all``
- # here because the docker tests only drive the container via
- # subprocess and don't import hermes_agent's optional deps.
- uv sync --locked --python 3.11 --extra dev
-
- - name: Run docker integration tests
- env:
- # Skip rebuild; use the image already loaded by the build step.
- HERMES_TEST_IMAGE: ${{ env.IMAGE_NAME }}:test
- # Match the policy in tests.yml :: test job — no accidental
- # real-API calls from inside the harness.
- OPENROUTER_API_KEY: ""
- OPENAI_API_KEY: ""
- NOUS_API_KEY: ""
- run: |
- scripts/run_tests.sh tests/docker/ --file-timeout 600
+ cache-from: ${{ matrix.cache-from }}
+ cache-to: ${{ (github.event_name != 'pull_request') && matrix.cache-to || '' }}
- name: Log in to Docker Hub
if: github.event_name == 'push' && github.ref == 'refs/heads/main' || github.event_name == 'release'
@@ -105,24 +69,24 @@ jobs:
username: ${{ secrets.DOCKERHUB_USERNAME }}
password: ${{ secrets.DOCKERHUB_TOKEN }}
- # Push amd64 by digest only (no tag). The merge job assembles the
+ # Push by digest only (no tag). The merge job assembles the
# tagged manifest list. `push-by-digest=true` is docker's recommended
# pattern for multi-runner multi-platform builds.
- - name: Push amd64 by digest
+ - name: Push ${{ matrix.arch }} by digest
id: push
if: github.event_name == 'push' && github.ref == 'refs/heads/main' || github.event_name == 'release'
uses: docker/build-push-action@bcafcacb16a39f128d818304e6c9c0c18556b85f # v7.1.0
with:
context: .
file: Dockerfile
- platforms: linux/amd64
+ platforms: ${{ matrix.platform }}
labels: |
org.opencontainers.image.revision=${{ github.sha }}
build-args: |
HERMES_GIT_SHA=${{ github.sha }}
outputs: type=image,name=${{ env.IMAGE_NAME }},push-by-digest=true,name-canonical=true,push=true
- cache-from: type=gha,scope=docker-amd64
- cache-to: type=gha,mode=max,scope=docker-amd64
+ cache-from: ${{ matrix.cache-from }}
+ cache-to: ${{ matrix.cache-to }}
# Write the digest to a file and upload it as an artifact so the
# merge job can stitch both per-arch digests into a manifest list.
@@ -137,121 +101,51 @@ jobs:
if: github.event_name == 'push' && github.ref == 'refs/heads/main' || github.event_name == 'release'
uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7
with:
- name: digest-amd64
+ name: digest-${{ matrix.arch }}
path: /tmp/digests/*
if-no-files-found: error
retention-days: 1
- # ---------------------------------------------------------------------------
- # Build, test, and optionally push the arm64 image.
- # ---------------------------------------------------------------------------
- build-arm64:
- if: github.repository == 'NousResearch/hermes-agent'
- runs-on: ubuntu-24.04-arm
- timeout-minutes: 45
- outputs:
- digest: ${{ steps.push.outputs.digest }}
- steps:
- - name: Checkout code
- uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2
-
- - name: Set up Docker Buildx
- uses: docker/setup-buildx-action@8d2750c68a42422c14e847fe6c8ac0403b4cbd6f # v3
-
- # Log in to ghcr.io so the registry-backed build cache below can be
- # read (cache-from) on every event and written (cache-to) on
- # push/release. Uses the workflow's GITHUB_TOKEN, which is valid for
- # the whole job — unlike the gha cache backend's short-lived Azure SAS
- # token, which expired mid-build on slow cold-cache arm64 runs and
- # crashed the build before the tests ran (the reason the gha cache
- # was removed from arm64 PRs in the first place).
- - name: Log in to ghcr.io (build cache)
- uses: docker/login-action@4907a6ddec9925e35a0a9e82d7399ccc52663121 # v4.1.0
- with:
- registry: ghcr.io
- username: ${{ github.actor }}
- password: ${{ secrets.GITHUB_TOKEN }}
-
- # Build once, load into the local daemon for testing, then push
- # by digest below. Reads AND writes the registry-backed cache so the
- # push reuses layers from this build and the next build starts warm.
+ # Run the docker-integration test suite against the freshly-built
+ # image already loaded into the local daemon (`:test`).
#
- # Registry cache (type=registry on ghcr.io) is used instead of the gha
- # cache that previously broke here: its credential is the job-lifetime
- # GITHUB_TOKEN, not a short-lived SAS token, so the cold-build-outlives-
- # token failure mode cannot recur.
- - name: Build image (arm64, cached publish)
- uses: docker/build-push-action@bcafcacb16a39f128d818304e6c9c0c18556b85f # v7.1.0
- with:
- context: .
- file: Dockerfile
- load: true
- platforms: linux/arm64
- tags: ${{ env.IMAGE_NAME }}:test
- build-args: |
- HERMES_GIT_SHA=${{ github.sha }}
- cache-from: type=registry,ref=ghcr.io/nousresearch/hermes-agent:buildcache-arm64
- cache-to: type=registry,ref=ghcr.io/nousresearch/hermes-agent:buildcache-arm64,mode=max
-
- - name: Install uv for docker tests
+ # Piggybacking here avoids a second image build: the build step
+ # already loaded the image into the daemon under
+ # `${IMAGE_NAME}:test`, so we just point ``HERMES_TEST_IMAGE`` at
+ # that. The fixture's ``HERMES_TEST_IMAGE`` branch (see
+ # tests/docker/conftest.py:62-63) short-circuits the rebuild.
+ #
+ # Why this job and not a standalone one: the image is 5GB+; passing
+ # it between jobs via ``docker save``/``upload-artifact`` is slower
+ # than the build itself. Reusing the existing daemon state is the
+ # cheapest path to coverage on every PR that touches docker code.
+ # ---------------------------------------------------------------------
+ - name: Install uv (for docker tests)
uses: astral-sh/setup-uv@fac544c07dec837d0ccb6301d7b5580bf5edae39 # 8.2.0
- - name: Set up Python 3.11 for docker tests
+ - name: Set up Python 3.11 (for docker tests)
run: uv python install 3.11
- - name: Install Python dependencies for docker tests
+ - name: Install Python dependencies (for docker tests)
run: |
+ # ``dev`` extra pulls in pytest, pytest-asyncio —
+ # everything tests/docker/ needs. We deliberately avoid ``all``
+ # here because the docker tests only drive the container via
+ # subprocess and don't import hermes_agent's optional deps.
uv sync --locked --python 3.11 --extra dev
- - name: Run docker tests
+ - name: Run docker integration tests
env:
# Skip rebuild; use the image already loaded by the build step.
HERMES_TEST_IMAGE: ${{ env.IMAGE_NAME }}:test
+ # Match the policy in tests.yml :: test job — no accidental
+ # real-API calls from inside the harness.
OPENROUTER_API_KEY: ""
OPENAI_API_KEY: ""
NOUS_API_KEY: ""
run: |
scripts/run_tests.sh tests/docker/ --file-timeout 600
- - name: Log in to Docker Hub
- if: github.event_name == 'push' && github.ref == 'refs/heads/main' || github.event_name == 'release'
- uses: docker/login-action@4907a6ddec9925e35a0a9e82d7399ccc52663121 # v4.1.0
- with:
- username: ${{ secrets.DOCKERHUB_USERNAME }}
- password: ${{ secrets.DOCKERHUB_TOKEN }}
-
- - name: Push arm64 by digest
- id: push
- if: github.event_name == 'push' && github.ref == 'refs/heads/main' || github.event_name == 'release'
- uses: docker/build-push-action@bcafcacb16a39f128d818304e6c9c0c18556b85f # v7.1.0
- with:
- context: .
- file: Dockerfile
- platforms: linux/arm64
- labels: |
- org.opencontainers.image.revision=${{ github.sha }}
- build-args: |
- HERMES_GIT_SHA=${{ github.sha }}
- outputs: type=image,name=${{ env.IMAGE_NAME }},push-by-digest=true,name-canonical=true,push=true
- cache-from: type=registry,ref=ghcr.io/nousresearch/hermes-agent:buildcache-arm64
- cache-to: type=registry,ref=ghcr.io/nousresearch/hermes-agent:buildcache-arm64,mode=max
-
- - name: Export digest
- if: github.event_name == 'push' && github.ref == 'refs/heads/main' || github.event_name == 'release'
- run: |
- mkdir -p /tmp/digests
- digest="${{ steps.push.outputs.digest }}"
- touch "/tmp/digests/${digest#sha256:}"
-
- - name: Upload digest artifact
- if: github.event_name == 'push' && github.ref == 'refs/heads/main' || github.event_name == 'release'
- uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7
- with:
- name: digest-arm64
- path: /tmp/digests/*
- if-no-files-found: error
- retention-days: 1
-
# ---------------------------------------------------------------------------
# Stitch both per-arch digests into a single tagged multi-arch manifest.
# This is a registry-side operation — no building, no layer re-push —
@@ -263,7 +157,7 @@ jobs:
merge:
if: github.repository == 'NousResearch/hermes-agent' && (github.event_name == 'push' && github.ref == 'refs/heads/main' || github.event_name == 'release')
runs-on: ubuntu-latest
- needs: [build-amd64, build-arm64]
+ needs: [build]
timeout-minutes: 10
steps:
- name: Download digests
diff --git a/.github/workflows/lint.yml b/.github/workflows/lint.yml
index 511119ca615f..fcee2c1b8e86 100644
--- a/.github/workflows/lint.yml
+++ b/.github/workflows/lint.yml
@@ -109,46 +109,6 @@ jobs:
--output .lint-reports/summary.md
cat .lint-reports/summary.md >> "$GITHUB_STEP_SUMMARY"
- - name: Upload reports as artifact
- uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7
- with:
- name: lint-reports
- path: .lint-reports/
- retention-days: 14
-
- - name: Post / update PR comment
- if: inputs.event_name == 'pull_request' && github.event.pull_request.head.repo.full_name == github.repository
- continue-on-error: true
- uses: actions/github-script@60a0d83039c74a4aee543508d2ffcb1c3799cdea # v7
- with:
- script: |
- const fs = require('fs');
- const body = fs.readFileSync('.lint-reports/summary.md', 'utf8');
- const marker = '';
- const fullBody = marker + '\n' + body;
-
- const { data: comments } = await github.rest.issues.listComments({
- owner: context.repo.owner,
- repo: context.repo.repo,
- issue_number: context.issue.number,
- });
- const existing = comments.find(c => c.body && c.body.includes(marker));
- if (existing) {
- await github.rest.issues.updateComment({
- owner: context.repo.owner,
- repo: context.repo.repo,
- comment_id: existing.id,
- body: fullBody,
- });
- } else {
- await github.rest.issues.createComment({
- owner: context.repo.owner,
- repo: context.repo.repo,
- issue_number: context.issue.number,
- body: fullBody,
- });
- }
-
ruff-blocking:
# Enforce the rules in pyproject.toml [tool.ruff.lint.select]. Currently
# PLW1514 (unspecified-encoding) — catches bare ``open()`` /
diff --git a/AGENTS.md b/AGENTS.md
index d8306d9bdb8a..e89c819844e6 100644
--- a/AGENTS.md
+++ b/AGENTS.md
@@ -1289,65 +1289,22 @@ scripts/run_tests.sh # full suite, CI-parity
scripts/run_tests.sh tests/gateway/ # one directory
scripts/run_tests.sh tests/agent/test_foo.py::test_x # one test
scripts/run_tests.sh -v --tb=long # pass-through pytest flags
-scripts/run_tests.sh --no-isolate tests/foo/ # disable subprocess isolation (faster, for debugging)
```
-### Subprocess-per-test isolation
+### Subprocess-per-test-file isolation
-Every test runs in a freshly-spawned Python subprocess via the in-tree plugin
-at `tests/_isolate_plugin.py`. This means module-level dicts/sets and
-ContextVars from one test cannot leak into the next — the historic
-`_reset_module_state` autouse fixture is gone.
+Every test file runs in a freshly-spawned Python subprocess via `run_tests_parallel.py`. This means module-level dicts/sets and
+ContextVars from one test file cannot leak into the next.
-Implementation notes:
+### Why the wrapper
-- The plugin uses `multiprocessing.get_context("spawn")`, which works on
- Linux, macOS, and Windows alike (POSIX `fork` is not used).
-- Per-test overhead is ~0.5–1.0s (Python startup + pytest collection). xdist
- parallelism amortizes this across cores; on a 20-core box the full suite
- finishes in roughly the same wall time as before, but flake-free.
-- `isolate_timeout` (configured in `pyproject.toml`) caps each test at 30s.
- Hangs are killed and surfaced as a failure report.
-- Pass `--no-isolate` to disable isolation — useful when debugging a single
- test interactively, or when you specifically want to verify state leakage.
-- The plugin disables itself in child processes (sentinel envvar
- `HERMES_ISOLATE_CHILD=1`), so there's no fork-bomb risk.
+| | Without wrapper | With wrapper |
+| ------------------- | ------------------------------------------- | ----------------------------------------- |
+| Provider API keys | Whatever is in your env (auto-detects pool) | All env vars except a specific few unset. |
+| HOME / `~/.hermes/` | Your real config+auth.json | Temp dir per test |
+| Timezone | Local TZ (PDT etc.) | UTC |
+| Locale | Whatever is set | C.UTF-8 |
-### Why the wrapper (and why the old "just call pytest" doesn't work)
-
-Five real sources of local-vs-CI drift the script closes:
-
-| | Without wrapper | With wrapper |
-|---|---|---|
-| Provider API keys | Whatever is in your env (auto-detects pool) | All `*_API_KEY`/`*_TOKEN`/etc. unset |
-| HOME / `~/.hermes/` | Your real config+auth.json | Temp dir per test |
-| Timezone | Local TZ (PDT etc.) | UTC |
-| Locale | Whatever is set | C.UTF-8 |
-| xdist workers | `-n auto` = all cores | `-n auto` (safe — subprocess isolation prevents cross-worker flakes) |
-
-`tests/conftest.py` also enforces points 1-4 as an autouse fixture so ANY pytest
-invocation (including IDE integrations) gets hermetic behavior — but the wrapper
-is belt-and-suspenders.
-
-### Running without the wrapper (only if you must)
-
-If you can't use the wrapper (e.g. inside an IDE that shells pytest directly),
-at minimum activate the venv. The isolation plugin loads automatically from
-`addopts` in `pyproject.toml`, so you get the same per-test process isolation
-either way.
-
-```bash
-source .venv/bin/activate # or: source venv/bin/activate
-python -m pytest tests/ -q
-```
-
-If you need to bypass isolation for fast feedback while debugging:
-
-```bash
-python -m pytest tests/agent/test_foo.py -q --no-isolate
-```
-
-Always run the full suite before pushing changes.
### Don't write change-detector tests
diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md
index 7f56b971d1e6..bad33481c745 100644
--- a/CONTRIBUTING.md
+++ b/CONTRIBUTING.md
@@ -149,13 +149,20 @@ this way, make sure you run the `hermes` entrypoint from this venv; running the
system `python3 -m hermes_cli.main` can pick up unrelated system Python
packages.
+Create the venv **outside** the cloned source tree. A venv that lives inside
+the directory the agent operates from can be wiped by a relative-path command
+the agent runs against its own checkout (`rm -rf venv`, `uv venv venv`, etc.),
+which silently destroys the running runtime mid-session. Keeping it outside the
+tree means no relative path from the workspace resolves to it.
+
```bash
git clone https://github.com/NousResearch/hermes-agent.git
cd hermes-agent
-# Create venv with Python 3.11
-uv venv venv --python 3.11
-export VIRTUAL_ENV="$(pwd)/venv"
+# Create venv with Python 3.11, OUTSIDE the source tree
+uv venv ~/.hermes/venvs/hermes-dev --python 3.11
+export VIRTUAL_ENV="$HOME/.hermes/venvs/hermes-dev"
+export PATH="$VIRTUAL_ENV/bin:$PATH"
# Install with all extras (messaging, cron, CLI menus, dev tools)
uv pip install -e ".[all,dev]"
diff --git a/README.md b/README.md
index 4caad13ce20e..ba1322a38920 100644
--- a/README.md
+++ b/README.md
@@ -232,10 +232,14 @@ scripts/run_tests.sh
Manual clone fallback (for throwaway clones/CI where you intentionally do not
want the managed install layout):
+Create the venv outside the cloned source tree — a venv inside the directory
+the agent operates from can be wiped by a relative-path command the agent runs
+against its own checkout, destroying the running runtime mid-session.
+
```bash
curl -LsSf https://astral.sh/uv/install.sh | sh
-uv venv .venv --python 3.11
-source .venv/bin/activate
+uv venv ~/.hermes/venvs/hermes-dev --python 3.11
+source ~/.hermes/venvs/hermes-dev/bin/activate
uv pip install -e ".[all,dev]"
scripts/run_tests.sh
```
diff --git a/acp_adapter/server.py b/acp_adapter/server.py
index a51db91d4e82..df773297346a 100644
--- a/acp_adapter/server.py
+++ b/acp_adapter/server.py
@@ -74,6 +74,10 @@
from acp_adapter.provenance import session_provenance_meta
from acp_adapter.session import SessionManager, SessionState, _expand_acp_enabled_toolsets
from acp_adapter.tools import build_tool_complete, build_tool_start
+from tools.approval import (
+ reset_hermes_interactive_context,
+ set_hermes_interactive_context,
+)
logger = logging.getLogger(__name__)
@@ -1446,20 +1450,23 @@ def stream_delta_cb(text: str) -> None:
# Approval callback is per-thread (thread-local, GHSA-qg5c-hvr5-hjgr).
# Set it INSIDE _run_agent so the TLS write happens in the executor
# thread — setting it here would write to the event-loop thread's TLS,
- # not the executor's. Also set HERMES_INTERACTIVE so approval.py
- # takes the CLI-interactive path (which calls the registered
- # callback via prompt_dangerous_approval) instead of the
- # non-interactive auto-approve branch (GHSA-96vc-wcxf-jjff).
+ # not the executor's. Interactive routing uses a contextvar in
+ # tools.approval (set_hermes_interactive_context) rather than
+ # os.environ["HERMES_INTERACTIVE"], so concurrent executor workers can't
+ # race on a process-global flag — one session's restore can't drop
+ # another onto the non-interactive auto-approve path mid-run
+ # (GHSA-96vc-wcxf-jjff). The contextvar write is isolated by the
+ # contextvars.copy_context() wrapper around the executor call below.
# ACP's conn.request_permission maps cleanly to the interactive
# callback shape — not the gateway-queue HERMES_EXEC_ASK path,
# which requires a notify_cb registered in _gateway_notify_cbs.
previous_approval_cb = None
- previous_interactive = None
+ interactive_token = None
edit_approval_token = None
previous_session_id = None
def _run_agent() -> dict:
- nonlocal previous_approval_cb, previous_interactive, edit_approval_token, previous_session_id
+ nonlocal previous_approval_cb, interactive_token, edit_approval_token, previous_session_id
# Bind HERMES_SESSION_KEY for this session so per-session caches
# (e.g. the interactive sudo password cache in tools.terminal_tool)
# scope to the ACP session rather than leaking across sessions
@@ -1491,9 +1498,10 @@ def _run_agent() -> dict:
except Exception:
logger.debug("Could not set ACP edit approval requester", exc_info=True)
# Signal to tools.approval that we have an interactive callback
- # and the non-interactive auto-approve path must not fire.
- previous_interactive = os.environ.get("HERMES_INTERACTIVE")
- os.environ["HERMES_INTERACTIVE"] = "1"
+ # and the non-interactive auto-approve path must not fire. Uses a
+ # contextvar (not os.environ) so concurrent executor workers don't
+ # race on the flag (GHSA-96vc-wcxf-jjff).
+ interactive_token = set_hermes_interactive_context(True)
# Propagate the originating ACP session id to tools that want to
# tag side-effects with it (e.g. ``kanban_create`` stamps it on
# the new task so clients can render a per-session board). Save
@@ -1513,11 +1521,9 @@ def _run_agent() -> dict:
logger.exception("Agent error in session %s", session_id)
return {"final_response": f"Error: {e}", "messages": state.history}
finally:
- # Restore HERMES_INTERACTIVE.
- if previous_interactive is None:
- os.environ.pop("HERMES_INTERACTIVE", None)
- else:
- os.environ["HERMES_INTERACTIVE"] = previous_interactive
+ # Restore the interactive contextvar for this context.
+ if interactive_token is not None:
+ reset_hermes_interactive_context(interactive_token)
# Restore HERMES_SESSION_ID symmetrically.
if previous_session_id is None:
os.environ.pop("HERMES_SESSION_ID", None)
diff --git a/agent/agent_init.py b/agent/agent_init.py
index 41f7cc11bbb1..dcfb1082d4c5 100644
--- a/agent/agent_init.py
+++ b/agent/agent_init.py
@@ -828,7 +828,7 @@ def _moa_reference_relay(event: str, **kwargs: Any) -> None:
client_kwargs["default_headers"] = build_nvidia_nim_headers(effective_base)
elif base_url_host_matches(effective_base, "api.routermint.com"):
client_kwargs["default_headers"] = _ra()._routermint_headers()
- elif base_url_host_matches(effective_base, "api.githubcopilot.com"):
+ elif base_url_host_matches(effective_base, "githubcopilot.com"):
from hermes_cli.models import copilot_default_headers
client_kwargs["default_headers"] = copilot_default_headers()
@@ -1665,6 +1665,12 @@ def _moa_reference_relay(event: str, **kwargs: Any) -> None:
abort_on_summary_failure=compression_abort_on_summary_failure,
max_tokens=agent.max_tokens,
)
+ _bind_session_state = getattr(agent.context_compressor, "bind_session_state", None)
+ if callable(_bind_session_state):
+ try:
+ _bind_session_state(session_db=session_db, session_id=agent.session_id)
+ except Exception:
+ pass
agent.compression_enabled = compression_enabled
agent.compression_in_place = compression_in_place
diff --git a/agent/agent_runtime_helpers.py b/agent/agent_runtime_helpers.py
index 21a14c977089..af64541a8285 100644
--- a/agent/agent_runtime_helpers.py
+++ b/agent/agent_runtime_helpers.py
@@ -368,6 +368,18 @@ def repair_message_sequence(agent, messages: List[Dict]) -> int:
host code) can feed in already-broken histories.
Repairs applied:
+ 0. Consecutive ``assistant`` messages with no intervening
+ ``tool``/``user`` turn — merged into a single assistant turn
+ (union of ``tool_calls``, concatenated ``content``). Strict
+ OpenAI-compatible providers (DeepSeek v4, Moonshot/Kimi) reject
+ a history where an ``assistant`` message carrying ``tool_calls``
+ is immediately followed by another ``assistant`` message instead
+ of its ``tool`` results — HTTP 400 "An assistant message with
+ 'tool_calls' must be followed by tool messages…". The split
+ shape is produced by recovery/continuation paths that append an
+ interim assistant turn (thinking-prefill, codex
+ incomplete-continuation) or by host-fed / legacy-persisted /
+ resumed histories. Refs #29148, #49147.
1. Stray ``tool`` messages whose ``tool_call_id`` doesn't match
any preceding assistant tool_call — dropped.
2. Consecutive ``user`` messages — merged with newline separator
@@ -387,12 +399,74 @@ def repair_message_sequence(agent, messages: List[Dict]) -> int:
repairs = 0
+ # Pass 0: merge consecutive assistant messages. Runs BEFORE Pass 1 so
+ # the merged turn's union of tool_call ids is known when Pass 1
+ # validates which tool-result messages are orphans. Two assistant
+ # messages are only adjacent here when nothing (no tool result, no
+ # user turn) separates them — an intervening ``tool`` message means
+ # two distinct, valid tool-call rounds that must NOT be merged.
+ #
+ # Codex Responses interim turns are exempt: the codex_responses
+ # api_mode legitimately keeps multiple consecutive incomplete
+ # assistant turns in history, each carrying its own encrypted
+ # continuation state (codex_reasoning_items / codex_message_items)
+ # that must be replayed verbatim. Collapsing them corrupts the
+ # Responses replay chain (the duplicate-detection logic at
+ # conversation_loop.py already de-dups identical codex interims).
+ def _is_codex_interim(m: Dict) -> bool:
+ return bool(
+ m.get("codex_reasoning_items")
+ or m.get("codex_message_items")
+ or m.get("finish_reason") == "incomplete"
+ )
+
+ collapsed: List[Dict] = []
+ for msg in messages:
+ if (
+ collapsed
+ and isinstance(msg, dict)
+ and msg.get("role") == "assistant"
+ and isinstance(collapsed[-1], dict)
+ and collapsed[-1].get("role") == "assistant"
+ and not _is_codex_interim(msg)
+ and not _is_codex_interim(collapsed[-1])
+ ):
+ prev = collapsed[-1]
+ # Union tool_calls (preserve order, both may carry them).
+ prev_calls = list(prev.get("tool_calls") or [])
+ new_calls = list(msg.get("tool_calls") or [])
+ if new_calls:
+ prev["tool_calls"] = prev_calls + new_calls
+ elif prev_calls:
+ prev["tool_calls"] = prev_calls
+ # Concatenate plain-text content; leave multimodal (list)
+ # content on either side alone to avoid mangling attachment
+ # blocks — fall back to keeping the existing content.
+ prev_content = prev.get("content")
+ new_content = msg.get("content")
+ if isinstance(prev_content, str) and isinstance(new_content, str):
+ joined = "\n".join(
+ p for p in (prev_content.strip(), new_content.strip()) if p
+ )
+ prev["content"] = joined
+ elif not prev_content and new_content is not None:
+ prev["content"] = new_content
+ # Carry reasoning_content from the later turn only if the
+ # earlier turn lacks it (strict thinking providers require a
+ # reasoning_content on the merged tool-call turn; the first
+ # non-empty one suffices).
+ if not prev.get("reasoning_content") and msg.get("reasoning_content"):
+ prev["reasoning_content"] = msg["reasoning_content"]
+ repairs += 1
+ continue
+ collapsed.append(msg)
+
# Pass 1: drop stray tool messages that don't follow a known
# assistant tool_call_id. Uses a rolling set of known ids refreshed
# on each assistant message.
known_tool_ids: set = set()
filtered: List[Dict] = []
- for msg in messages:
+ for msg in collapsed:
if not isinstance(msg, dict):
filtered.append(msg)
continue
@@ -663,6 +737,25 @@ def recover_with_credential_pool(
elif status_code in {401, 403}:
effective_reason = FailoverReason.auth
+ if effective_reason == FailoverReason.upstream_rate_limit:
+ # An upstream provider (e.g. DeepSeek behind OpenRouter) is
+ # rate-limiting the aggregator's traffic — the user's credential is
+ # healthy. Do NOT rotate or mark exhausted; let the caller's fallback
+ # path switch to a different model entirely.
+ upstream = (error_context or {}).get("upstream_provider") if error_context else None
+ if upstream:
+ _ra().logger.info(
+ "Upstream provider %s rate-limited via aggregator — skipping "
+ "credential rotation, deferring to fallback chain",
+ upstream,
+ )
+ else:
+ _ra().logger.info(
+ "Upstream aggregator 429 (provider unknown) — skipping "
+ "credential rotation, deferring to fallback chain"
+ )
+ return False, has_retried_429
+
if effective_reason == FailoverReason.billing:
rotate_status = status_code if status_code is not None else 402
next_entry = pool.mark_exhausted_and_rotate(status_code=rotate_status, error_context=error_context)
@@ -1281,7 +1374,11 @@ def dump_api_request_debug(
dump_payload["error"] = error_info
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S_%f")
- dump_file = agent.logs_dir / f"request_dump_{agent.session_id}_{timestamp}.json"
+ # Sanitize the session ID into a traversal-free path segment — it can
+ # originate from untrusted input (X-Hermes-Session-Id header), and an
+ # unsanitized "../"-shaped ID would write the dump outside logs_dir.
+ safe_sid = _ra()._safe_session_filename_component(agent.session_id)
+ dump_file = agent.logs_dir / f"request_dump_{safe_sid}_{timestamp}.json"
# Redact secrets before persisting/printing. This dump captures the
# full request body (system prompt, tool defs, context-embedded
@@ -1621,6 +1718,18 @@ def switch_model(agent, new_model, new_provider, api_key='', base_url='', api_mo
if (new_provider or "").strip().lower() == "moa":
from agent.moa_loop import MoAClient
+ # The MoA virtual provider speaks only chat.completions via the
+ # MoAClient facade — the aggregator's real transport
+ # (codex_responses / anthropic_messages) is resolved and applied
+ # *inside* the reference/aggregator fan-out, never on the outer
+ # primary call. determine_api_mode("moa", ...) above may have left
+ # api_mode set to the aggregator's transport; if the conversation
+ # loop sees that, it dispatches client.responses.create (which the
+ # facade has no .responses for) and the call falls through to the
+ # moa://local placeholder → HTTP 404 → fallback to a reference
+ # model. Pin chat_completions here so the primary call always goes
+ # through MoAClient.chat.completions, matching agent_init.py.
+ agent.api_mode = "chat_completions"
agent.api_key = api_key or "moa-virtual-provider"
agent.base_url = "moa://local"
agent._client_kwargs = {}
@@ -2159,7 +2268,7 @@ def sanitize_api_messages(messages: List[Dict[str, Any]]) -> List[Dict[str, Any]
result_call_ids: set = set()
for msg in messages:
if msg.get("role") == "tool":
- cid = msg.get("tool_call_id")
+ cid = (msg.get("tool_call_id") or "").strip()
if cid:
result_call_ids.add(cid)
@@ -2168,7 +2277,7 @@ def sanitize_api_messages(messages: List[Dict[str, Any]]) -> List[Dict[str, Any]
if orphaned_results:
messages = [
m for m in messages
- if not (m.get("role") == "tool" and m.get("tool_call_id") in orphaned_results)
+ if not (m.get("role") == "tool" and (m.get("tool_call_id") or "").strip() in orphaned_results)
]
_ra().logger.debug(
"Pre-call sanitizer: removed %d orphaned tool result(s)",
@@ -2202,7 +2311,7 @@ def sanitize_api_messages(messages: List[Dict[str, Any]]) -> List[Dict[str, Any]
def looks_like_codex_intermediate_ack(
agent,
- user_message: str,
+ user_message: Any,
assistant_content: str,
messages: List[Dict[str, Any]],
require_workspace: bool = True,
@@ -2282,7 +2391,14 @@ def looks_like_codex_intermediate_ack(
if not require_workspace:
return True
- user_text = (user_message or "").strip().lower()
+ # ``user_message`` is typed ``str`` but can arrive as an OpenAI-style
+ # multi-part content list (``[{type:"text",...}, {type:"image_url",...}]``)
+ # for vision requests routed through the OpenAI-compat API server. A
+ # truthy list survives ``(user_message or "")`` and then ``.strip()``
+ # raises ``AttributeError`` — flatten to text first.
+ from agent.codex_responses_adapter import _summarize_user_message_for_log
+
+ user_text = _summarize_user_message_for_log(user_message).strip().lower()
user_targets_workspace = (
any(marker in user_text for marker in workspace_markers)
or "~/" in user_text
diff --git a/agent/auxiliary_client.py b/agent/auxiliary_client.py
index dfeec87e12d3..c24cc972a2e5 100644
--- a/agent/auxiliary_client.py
+++ b/agent/auxiliary_client.py
@@ -124,6 +124,15 @@ def _openai_http_client_kwargs(
def _create_openai_client(*, api_key: str, base_url: str, **kwargs: Any) -> Any:
kwargs = {**_openai_http_client_kwargs(base_url), **kwargs}
+ # Hermes owns auxiliary retry + provider/model fallback policy (the
+ # same-provider transient retry in call_llm plus the except-chain
+ # fallback). The OpenAI SDK's own default (max_retries=2 → up to 3
+ # attempts) silently multiplies the effective wall time of every aux call
+ # by 3× on a slow/hung endpoint, so a 120s timeout can stall ~360s before
+ # Hermes sees a single failure (issue #54465). Disable SDK-internal retries
+ # by default and let Hermes control the budget; explicit callers can still
+ # override via kwargs.
+ kwargs.setdefault("max_retries", 0)
return OpenAI(api_key=api_key, base_url=base_url, **kwargs)
@@ -1615,7 +1624,7 @@ def _resolve_api_key_provider() -> Tuple[Optional[OpenAI], Optional[str]]:
extra = {}
if base_url_host_matches(base_url, "api.kimi.com"):
extra["default_headers"] = {"User-Agent": "claude-code/0.1.0"}
- elif base_url_host_matches(base_url, "api.githubcopilot.com"):
+ elif base_url_host_matches(base_url, "githubcopilot.com"):
from hermes_cli.models import copilot_default_headers
extra["default_headers"] = copilot_default_headers()
@@ -1655,7 +1664,7 @@ def _resolve_api_key_provider() -> Tuple[Optional[OpenAI], Optional[str]]:
extra = {}
if base_url_host_matches(base_url, "api.kimi.com"):
extra["default_headers"] = {"User-Agent": "claude-code/0.1.0"}
- elif base_url_host_matches(base_url, "api.githubcopilot.com"):
+ elif base_url_host_matches(base_url, "githubcopilot.com"):
from hermes_cli.models import copilot_default_headers
extra["default_headers"] = copilot_default_headers()
@@ -2590,6 +2599,27 @@ def _is_rate_limit_error(exc: Exception) -> bool:
return False
+def _is_timeout_error(exc: Exception) -> bool:
+ """Detect a request timeout — the full-budget stall, distinct from a fast
+ connection drop.
+
+ A timeout burns the entire configured ``timeout`` before surfacing, so a
+ same-provider retry on the critical compression path doubles the
+ user-visible wall time (issue #54465). A streaming-close / dropped
+ connection, by contrast, fails fast and is cheap to retry — those stay on
+ the retry path even for compression.
+ """
+ try:
+ from openai import APITimeoutError
+ if isinstance(exc, APITimeoutError):
+ return True
+ except ImportError:
+ pass
+ if "Timeout" in type(exc).__name__:
+ return True
+ return "timed out" in str(exc).lower()
+
+
def _is_connection_error(exc: Exception) -> bool:
"""Detect connection/network errors that warrant provider fallback.
@@ -2924,7 +2954,7 @@ def _recoverable_pool_provider(
return "nous"
if base_url_host_matches(base, "api.anthropic.com"):
return "anthropic"
- if base_url_host_matches(base, "api.githubcopilot.com"):
+ if base_url_host_matches(base, "githubcopilot.com"):
return "copilot"
if base_url_host_matches(base, "api.kimi.com"):
return "kimi-coding"
@@ -3793,7 +3823,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"] = build_or_headers()
- elif base_url_host_matches(sync_base_url, "api.githubcopilot.com"):
+ elif base_url_host_matches(sync_base_url, "githubcopilot.com"):
from hermes_cli.copilot_auth import copilot_request_headers
async_kwargs["default_headers"] = copilot_request_headers(
@@ -3824,6 +3854,9 @@ def _to_async_client(sync_client, model: str, is_vision: bool = False):
**_openai_http_client_kwargs(sync_base_url, async_mode=True),
**async_kwargs,
}
+ # See _create_openai_client: disable SDK-internal retries so Hermes owns
+ # the auxiliary retry/timeout budget (issue #54465).
+ async_kwargs.setdefault("max_retries", 0)
return AsyncOpenAI(**async_kwargs), model
@@ -4095,7 +4128,7 @@ def _wrap_if_needed(client_obj, final_model_str: str, base_url_str: str = "",
extra["default_query"] = _dq
if base_url_host_matches(custom_base, "api.kimi.com"):
extra["default_headers"] = {"User-Agent": "claude-code/0.1.0"}
- elif base_url_host_matches(custom_base, "api.githubcopilot.com"):
+ elif base_url_host_matches(custom_base, "githubcopilot.com"):
from hermes_cli.copilot_auth import copilot_request_headers
extra["default_headers"] = copilot_request_headers(
is_agent_turn=True, is_vision=is_vision
@@ -4348,7 +4381,7 @@ def _wrap_if_needed(client_obj, final_model_str: str, base_url_str: str = "",
headers = {}
if base_url_host_matches(base_url, "api.kimi.com"):
headers["User-Agent"] = "claude-code/0.1.0"
- elif base_url_host_matches(base_url, "api.githubcopilot.com"):
+ elif base_url_host_matches(base_url, "githubcopilot.com"):
from hermes_cli.copilot_auth import copilot_request_headers
headers.update(copilot_request_headers(
@@ -4821,9 +4854,14 @@ def auxiliary_max_tokens_param(value: int, *, model: Optional[str] = None) -> di
or_key = os.getenv("OPENROUTER_API_KEY")
# Use max_completion_tokens for direct OpenAI-compatible providers that reject
# max_tokens on newer GPT-4o/o-series/GPT-5-style models.
+ _custom_host = base_url_hostname(custom_base) or ""
if (not or_key
and _read_nous_auth() is None
- and base_url_hostname(custom_base) in {"api.openai.com", "api.githubcopilot.com"}):
+ and (
+ _custom_host == "api.openai.com"
+ or _custom_host == "api.githubcopilot.com"
+ or _custom_host.endswith(".githubcopilot.com")
+ )):
return {"max_completion_tokens": value}
# ...and for any caller serving a newer OpenAI-family model by name.
if model_forces_max_completion_tokens(model):
@@ -5200,9 +5238,10 @@ def _resolve_task_provider_model(
3. "auto" (full auto-detection chain)
Returns (provider, model, base_url, api_key, api_mode) where model may
- be None (use provider default). When base_url is set, provider is forced
- to "custom" and the task uses that direct endpoint. api_mode is one of
- "chat_completions", "codex_responses", or None (auto-detect).
+ be None (use provider default). A bare base_url is treated as custom, but
+ a first-class provider plus base_url keeps the provider identity so its
+ auth, transport, and request-shaping behavior still apply. api_mode is one
+ of "chat_completions", "codex_responses", or None (auto-detect).
"""
cfg_provider = None
cfg_model = None
@@ -5235,11 +5274,35 @@ def _expand_direct_api_alias(prov: Optional[str], existing_base: Optional[str])
return prov, existing_base
return "custom", existing_base or target_base
+ def _preserve_provider_with_base_url(prov: Optional[str]) -> bool:
+ normalized = str(prov or "").strip().lower()
+ if normalized in {"", "auto", "custom"} or normalized.startswith("custom:"):
+ return False
+ try:
+ from hermes_cli.providers import get_provider
+
+ return get_provider(normalized) is not None
+ except Exception:
+ # Keep the high-risk provider-backed routes safe even if provider
+ # catalog loading is unavailable during early import/test paths.
+ return normalized in {
+ "anthropic",
+ "copilot",
+ "copilot-acp",
+ "minimax-oauth",
+ "nous",
+ "openai-codex",
+ "qwen-oauth",
+ "xai-oauth",
+ }
+
if provider:
provider, base_url = _expand_direct_api_alias(provider, base_url)
if cfg_provider:
cfg_provider, cfg_base_url = _expand_direct_api_alias(cfg_provider, cfg_base_url)
+ if base_url and _preserve_provider_with_base_url(provider):
+ return provider, resolved_model, base_url, api_key, resolved_api_mode
if base_url:
return "custom", resolved_model, base_url, api_key, resolved_api_mode
if provider:
@@ -5489,10 +5552,24 @@ def _build_call_kwargs(
# ``/anthropic`` endpoint reached through the OpenAI SDK wrapper), where
# max_tokens is a MANDATORY field — omitting it is a hard 400. Keep it only
# there.
+ #
+ # NVIDIA NIM (integrate.api.nvidia.com and local NIM endpoints) is a
+ # second exception: some models—notably minimaxai/minimax-m3—return HTTP
+ # 200 with an empty choices[] payload when max_tokens is omitted. The main
+ # NVIDIA chat path already sends an output cap via the provider profile;
+ # preserve it on the auxiliary path too.
_effective_base = base_url or (
_current_custom_base_url() if provider == "custom" else ""
)
- if _is_anthropic_compat_endpoint(provider, _effective_base):
+ _provider_norm = str(provider or "").strip().lower()
+ _is_nvidia_nim = (
+ _provider_norm in {"nvidia", "nvidia-nim", "nim", "build-nvidia", "nemotron"}
+ or base_url_host_matches(_effective_base, "integrate.api.nvidia.com")
+ )
+ if (
+ _is_anthropic_compat_endpoint(provider, _effective_base)
+ or _is_nvidia_nim
+ ):
kwargs["max_tokens"] = max_tokens
if tools:
@@ -5633,6 +5710,9 @@ def call_llm(
tools: list = None,
timeout: float = None,
extra_body: dict = None,
+ api_mode: str = None,
+ stream: bool = False,
+ stream_options: dict = None,
) -> Any:
"""Centralized synchronous LLM call.
@@ -5645,21 +5725,32 @@ def call_llm(
Reads provider:model from config/env. Ignored if provider is set.
provider: Explicit provider override.
model: Explicit model override.
+ api_mode: Explicit API mode override (e.g. "codex_responses",
+ "anthropic_messages"). Takes precedence over task config.
messages: Chat messages list.
temperature: Sampling temperature (None = provider default).
max_tokens: Max output tokens (handles max_tokens vs max_completion_tokens).
tools: Tool definitions (for function calling).
timeout: Request timeout in seconds (None = read from auxiliary.{task}.timeout config).
extra_body: Additional request body fields.
+ stream: When True, return the raw SDK streaming iterator instead of a
+ validated complete response. The caller is responsible for consuming
+ chunks (and for any fallback). Used by the MoA aggregator so its
+ output can stream to the user.
+ stream_options: Passed through to the request when stream is True
+ (e.g. {"include_usage": True}).
Returns:
- Response object with .choices[0].message.content
+ Response object with .choices[0].message.content, OR — when stream=True —
+ the raw streaming iterator from client.chat.completions.create().
Raises:
RuntimeError: If no provider is configured.
"""
resolved_provider, resolved_model, resolved_base_url, resolved_api_key, resolved_api_mode = _resolve_task_provider_model(
task, provider, model, base_url, api_key)
+ if api_mode:
+ resolved_api_mode = api_mode
effective_extra_body = _get_task_extra_body(task)
effective_extra_body.update(extra_body or {})
@@ -5753,6 +5844,20 @@ def call_llm(
if _is_anthropic_compat_endpoint(resolved_provider, _client_base):
kwargs["messages"] = _convert_openai_images_to_anthropic(kwargs["messages"])
+ # Streaming path: return the raw SDK Stream iterator directly. This is used by
+ # the MoA aggregator so its tokens stream to the user. It deliberately skips
+ # _validate_llm_response and the temperature/max_tokens/payment fallback chain
+ # below — those all assume a complete response object, whereas a stream is
+ # consumed chunk-by-chunk by the caller. The caller (the agent's streaming
+ # consumer) owns chunk reassembly, stale-stream detection, and falling back to
+ # a non-streaming call on error. stream_options is best-effort: providers that
+ # reject it surface an error the caller's fallback already handles.
+ if stream:
+ kwargs["stream"] = True
+ if stream_options:
+ kwargs["stream_options"] = stream_options
+ return client.chat.completions.create(**kwargs)
+
# Handle unsupported temperature, max_tokens vs max_completion_tokens retry,
# then payment fallback.
try:
@@ -5771,6 +5876,21 @@ def call_llm(
except Exception as transient_err:
if not _is_transient_transport_error(transient_err):
raise
+ # Compression is on the critical preflight path: a user cannot
+ # continue or resume an oversized session until it compacts. A
+ # same-provider retry on a timeout means another full ``timeout``-
+ # long wall-clock block before the except-chain below can fall
+ # back — doubling the user-visible stall (issue #54465). Skip the
+ # same-provider retry for compression on a full-budget timeout and
+ # fall straight through to provider/model fallback; fast blips (a
+ # streaming-close or a 5xx) still retry, since those are cheap.
+ if task == "compression" and _is_timeout_error(transient_err):
+ logger.info(
+ "Auxiliary compression: timeout on the critical path; "
+ "skipping same-provider retry and falling back: %s",
+ transient_err,
+ )
+ raise
logger.info(
"Auxiliary %s: transient transport error; retrying once on "
"the same provider before fallback: %s",
@@ -6296,6 +6416,16 @@ async def async_call_llm(
except Exception as transient_err:
if not _is_transient_transport_error(transient_err):
raise
+ # See call_llm(): compression is on the critical preflight path,
+ # so skip the same-provider retry on a full-budget timeout and
+ # fall straight through to fallback (issue #54465).
+ if task == "compression" and _is_timeout_error(transient_err):
+ logger.info(
+ "Auxiliary compression (async): timeout on the critical "
+ "path; skipping same-provider retry and falling back: %s",
+ transient_err,
+ )
+ raise
logger.info(
"Auxiliary %s (async): transient transport error; retrying "
"once on the same provider before fallback: %s",
diff --git a/agent/chat_completion_helpers.py b/agent/chat_completion_helpers.py
index 7a5e75347237..aada15f51ed5 100644
--- a/agent/chat_completion_helpers.py
+++ b/agent/chat_completion_helpers.py
@@ -632,7 +632,7 @@ def build_api_kwargs(agent, api_messages: list) -> dict:
_ct = agent._get_transport()
is_github_responses = (
base_url_host_matches(agent.base_url, "models.github.ai")
- or base_url_host_matches(agent.base_url, "api.githubcopilot.com")
+ or base_url_host_matches(agent.base_url, "githubcopilot.com")
)
is_codex_backend = (
agent.provider == "openai-codex"
@@ -702,7 +702,7 @@ def build_api_kwargs(agent, api_messages: list) -> dict:
_is_or = agent._is_openrouter_url()
_is_gh = (
base_url_host_matches(agent._base_url_lower, "models.github.ai")
- or base_url_host_matches(agent._base_url_lower, "api.githubcopilot.com")
+ or base_url_host_matches(agent._base_url_lower, "githubcopilot.com")
)
_is_nous = "nousresearch" in agent._base_url_lower
_is_nvidia = "integrate.api.nvidia.com" in agent._base_url_lower
@@ -1124,7 +1124,7 @@ def try_activate_fallback(agent, reason: "FailoverReason | None" = None) -> bool
auth resolution and client construction — no duplicated provider→key
mappings.
"""
- if reason in {FailoverReason.rate_limit, FailoverReason.billing}:
+ if reason in {FailoverReason.rate_limit, FailoverReason.billing, FailoverReason.upstream_rate_limit}:
# Only start cooldown when leaving the primary provider. If we're
# already on a fallback and chain-switching, the primary wasn't the
# source of the 429 so the cooldown should not be reset/extended.
@@ -1142,7 +1142,7 @@ def try_activate_fallback(agent, reason: "FailoverReason | None" = None) -> bool
# provider again. Guards the cross-turn replay storm in #24996.
if (
len(agent._fallback_chain) > 0
- and reason not in {FailoverReason.rate_limit, FailoverReason.billing}
+ and reason not in {FailoverReason.rate_limit, FailoverReason.billing, FailoverReason.upstream_rate_limit}
):
_existing_cooldown = getattr(agent, "_rate_limited_until", 0) or 0
agent._rate_limited_until = max(
@@ -2086,7 +2086,7 @@ def _call_chat_completions():
entry["function"]["arguments"] += tc_delta.function.arguments
extra = getattr(tc_delta, "extra_content", None)
if extra is None and hasattr(tc_delta, "model_extra"):
- extra = (tc_delta.model_extra or {}).get("extra_content")
+ extra = (tc_delta.model_extra if isinstance(tc_delta.model_extra, dict) else {}).get("extra_content")
if extra is not None:
if hasattr(extra, "model_dump"):
extra = extra.model_dump()
diff --git a/agent/coding_context.py b/agent/coding_context.py
index 8fb51a0b04d6..00f6d996d478 100644
--- a/agent/coding_context.py
+++ b/agent/coding_context.py
@@ -353,6 +353,29 @@ def _coding_mode(config: Optional[dict[str, Any]]) -> str:
return "auto"
+def _coding_instructions(config: Optional[dict[str, Any]]) -> str:
+ """Standing operator instructions for the coding posture (config).
+
+ ``agent.coding_instructions`` — a string or list of strings appended to the
+ coding brief as an extra stable system block, so a user can pin project-wide
+ coding-workflow rules (e.g. "for UI work don't run tsc/lint until I approve;
+ clean the diff before committing") without editing the shipped brief.
+ Cache-safe: resolved once per session into the stable system-prompt tier,
+ like the rest of the posture.
+ """
+ if config is None:
+ try:
+ from hermes_cli.config import load_config
+
+ config = load_config()
+ except Exception:
+ config = {}
+ raw = ((config or {}).get("agent", {}) or {}).get("coding_instructions", "")
+ if isinstance(raw, (list, tuple)):
+ return "\n".join(str(item).strip() for item in raw if str(item).strip())
+ return str(raw or "").strip()
+
+
def _resolve_cwd(cwd: Optional[str | Path]) -> Path:
if cwd:
return Path(cwd).expanduser()
@@ -459,6 +482,9 @@ class RuntimeMode:
# only to steer edit-format guidance toward the model's family — see
# ``_edit_format_line``. Fixed for the session, so cache-safe.
model: Optional[str] = None
+ # Standing operator instructions (``agent.coding_instructions``), appended
+ # as an extra stable system block. Empty unless the user configures it.
+ instructions: str = ""
@property
def kind(self) -> str:
@@ -505,6 +531,10 @@ def system_blocks(self) -> list[str]:
workspace = build_coding_workspace_block(self.cwd)
if workspace:
blocks.append(workspace)
+ # Operator instructions ride their own block so the brief (block 0) stays
+ # byte-stable and cache-keyed independently of user config.
+ if self.instructions:
+ blocks.append(f"Operator instructions (from config):\n{self.instructions}")
return blocks
def compact_skill_categories(self) -> frozenset[str]:
@@ -557,6 +587,7 @@ def resolve_runtime_mode(
cwd=resolved_cwd,
config_mode=mode,
model=model,
+ instructions=_coding_instructions(config),
)
diff --git a/agent/context_breakdown.py b/agent/context_breakdown.py
new file mode 100644
index 000000000000..0e2eb772f2ff
--- /dev/null
+++ b/agent/context_breakdown.py
@@ -0,0 +1,156 @@
+"""Live session context-window breakdown for UI surfaces.
+
+Estimates how the next provider request is composed: system prompt tiers,
+tool schemas, and conversation history. Uses the same rough char/4 heuristic
+as ``agent.model_metadata.estimate_request_tokens_rough`` so numbers align
+with compression thresholds — not exact tokenizer counts.
+"""
+
+from __future__ import annotations
+
+import json
+import re
+from typing import Any, Dict, List, Optional, Sequence, Tuple
+
+_SKILLS_BLOCK_RE = re.compile(r".*?", re.DOTALL)
+
+_SUBAGENT_TOOL_NAMES = frozenset({"delegate_task"})
+
+_CATEGORY_COLORS = {
+ "system_prompt": "var(--context-usage-system)",
+ "tool_definitions": "var(--context-usage-tools)",
+ "rules": "var(--context-usage-rules)",
+ "skills": "var(--context-usage-skills)",
+ "mcp": "var(--context-usage-mcp)",
+ "subagent_definitions": "var(--context-usage-subagents)",
+ "memory": "var(--context-usage-memory)",
+ "conversation": "var(--context-usage-conversation)",
+}
+
+
+def _chars_to_tokens(text: str) -> int:
+ if not text:
+ return 0
+ return (len(text) + 3) // 4
+
+
+def _json_tokens(value: Any) -> int:
+ if not value:
+ return 0
+ return _chars_to_tokens(json.dumps(value, ensure_ascii=False))
+
+
+def _tool_name(tool: dict) -> str:
+ fn = tool.get("function") if isinstance(tool, dict) else None
+ if isinstance(fn, dict):
+ return str(fn.get("name") or "")
+ return str(tool.get("name") or "")
+
+
+def _split_tools(tools: Sequence[dict]) -> Tuple[List[dict], List[dict], List[dict]]:
+ builtin: List[dict] = []
+ mcp: List[dict] = []
+ subagent: List[dict] = []
+ for tool in tools:
+ name = _tool_name(tool)
+ if name.startswith("mcp_"):
+ mcp.append(tool)
+ elif name in _SUBAGENT_TOOL_NAMES:
+ subagent.append(tool)
+ else:
+ builtin.append(tool)
+ return builtin, mcp, subagent
+
+
+def _memory_blocks(agent: Any) -> Tuple[str, str]:
+ memory_block = ""
+ user_block = ""
+ store = getattr(agent, "_memory_store", None)
+ if store is None:
+ return memory_block, user_block
+ try:
+ if getattr(agent, "_memory_enabled", True):
+ memory_block = store.format_for_system_prompt("memory") or ""
+ if getattr(agent, "_user_profile_enabled", True):
+ user_block = store.format_for_system_prompt("user") or ""
+ except Exception:
+ pass
+ return memory_block, user_block
+
+
+def _strip_blocks(text: str, *blocks: str) -> str:
+ out = text
+ for block in blocks:
+ if block:
+ out = out.replace(block, "")
+ return out.strip()
+
+
+def compute_session_context_breakdown(
+ agent: Any,
+ messages: Optional[List[dict]] = None,
+) -> Dict[str, Any]:
+ """Return a Cursor-style context usage breakdown for one live agent."""
+ from agent.model_metadata import estimate_messages_tokens_rough
+ from agent.system_prompt import build_system_prompt_parts
+
+ parts = build_system_prompt_parts(agent)
+ stable = parts.get("stable", "") or ""
+ context = parts.get("context", "") or ""
+ volatile = parts.get("volatile", "") or ""
+
+ skills_match = _SKILLS_BLOCK_RE.search(stable)
+ skills_index = skills_match.group(0) if skills_match else ""
+
+ memory_block, user_block = _memory_blocks(agent)
+ memory_text = "\n\n".join(part for part in (memory_block, user_block) if part).strip()
+
+ system_core = _strip_blocks(stable, skills_index)
+ system_tail = _strip_blocks(volatile, memory_block, user_block)
+ system_prompt_text = "\n\n".join(part for part in (system_core, system_tail) if part).strip()
+
+ tools = list(getattr(agent, "tools", None) or [])
+ builtin_tools, mcp_tools, subagent_tools = _split_tools(tools)
+
+ conversation_tokens = estimate_messages_tokens_rough(messages or [])
+
+ categories = [
+ ("system_prompt", "System prompt", _chars_to_tokens(system_prompt_text)),
+ ("tool_definitions", "Tool definitions", _json_tokens(builtin_tools)),
+ ("rules", "Rules", _chars_to_tokens(context)),
+ ("skills", "Skills", _chars_to_tokens(skills_index)),
+ ("mcp", "MCP", _json_tokens(mcp_tools)),
+ ("subagent_definitions", "Subagent definitions", _json_tokens(subagent_tools)),
+ ("memory", "Memory", _chars_to_tokens(memory_text)),
+ ("conversation", "Conversation", conversation_tokens),
+ ]
+
+ estimated_total = sum(tokens for _, _, tokens in categories)
+
+ comp = getattr(agent, "context_compressor", None)
+ context_max = int(getattr(comp, "context_length", 0) or 0) if comp else 0
+ measured_used = int(getattr(comp, "last_prompt_tokens", 0) or 0) if comp else 0
+ context_used = measured_used if measured_used > 0 else estimated_total
+ context_percent = (
+ max(0, min(100, round(context_used / context_max * 100)))
+ if context_max
+ else 0
+ )
+
+ return {
+ "categories": [
+ {
+ "color": _CATEGORY_COLORS.get(category_id, "var(--ui-text-tertiary)"),
+ "id": category_id,
+ "label": label,
+ "tokens": tokens,
+ }
+ for category_id, label, tokens in categories
+ if tokens > 0
+ ],
+ "context_max": context_max,
+ "context_percent": context_percent,
+ "context_used": context_used,
+ "estimated_total": estimated_total,
+ "model": getattr(agent, "model", "") or "",
+ }
diff --git a/agent/context_compressor.py b/agent/context_compressor.py
index fbde99bda5f9..4bccda13808b 100644
--- a/agent/context_compressor.py
+++ b/agent/context_compressor.py
@@ -19,6 +19,7 @@
import hashlib
import json
import logging
+import sqlite3
import re
import time
from typing import Any, Dict, List, Optional
@@ -638,6 +639,7 @@ def on_session_reset(self) -> 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
+ self._last_summary_error = None
self.last_real_prompt_tokens = 0
self.last_compression_rough_tokens = 0
self.last_rough_tokens_when_real_prompt_fit = 0
@@ -659,6 +661,104 @@ def on_session_end(self, session_id: str, messages: List[Dict[str, Any]]) -> Non
"""
self._previous_summary = None
+ def bind_session_state(self, session_db: Any = None, session_id: str = "") -> None:
+ """Bind the current session row so durable cooldowns can round-trip."""
+ self._session_db = session_db
+ self._session_id = session_id or ""
+ self._summary_failure_cooldown_until = 0.0
+ self._last_summary_error = None
+ self.get_active_compression_failure_cooldown()
+
+ def on_session_start(self, session_id: str, **kwargs) -> None:
+ """Bind session-scoped compression state for a new or resumed session."""
+ super().on_session_start(session_id, **kwargs)
+ self.bind_session_state(kwargs.get("session_db", getattr(self, "_session_db", None)), session_id)
+
+ def get_active_compression_failure_cooldown(self) -> Optional[Dict[str, Any]]:
+ """Return the live compression-failure cooldown for the bound session."""
+ now_mono = time.monotonic()
+ if self._summary_failure_cooldown_until > now_mono:
+ return {
+ "cooldown_until": time.time() + (
+ self._summary_failure_cooldown_until - now_mono
+ ),
+ "remaining_seconds": self._summary_failure_cooldown_until - now_mono,
+ "error": self._last_summary_error,
+ }
+
+ session_db = getattr(self, "_session_db", None)
+ session_id = getattr(self, "_session_id", "")
+ if not session_db or not session_id:
+ return None
+
+ getter = getattr(session_db, "get_compression_failure_cooldown", None)
+ if getter is None:
+ return None
+ try:
+ state = getter(session_id)
+ except sqlite3.Error as exc:
+ logger.debug("compression failure cooldown lookup failed: %s", exc)
+ return None
+ except Exception:
+ return None
+ if not state:
+ return None
+
+ remaining_seconds = float(state.get("remaining_seconds") or 0.0)
+ if remaining_seconds <= 0:
+ return None
+
+ self._summary_failure_cooldown_until = now_mono + remaining_seconds
+ self._last_summary_error = state.get("error")
+ return {
+ "cooldown_until": float(state.get("cooldown_until") or 0.0),
+ "remaining_seconds": remaining_seconds,
+ "error": self._last_summary_error,
+ }
+
+ def _record_compression_failure_cooldown(
+ self,
+ cooldown_seconds: float,
+ error: Optional[str],
+ ) -> None:
+ cooldown_until = time.time() + cooldown_seconds
+ self._summary_failure_cooldown_until = time.monotonic() + cooldown_seconds
+ self._last_summary_error = error
+
+ session_db = getattr(self, "_session_db", None)
+ session_id = getattr(self, "_session_id", "")
+ if not session_db or not session_id:
+ return
+
+ recorder = getattr(session_db, "record_compression_failure_cooldown", None)
+ if recorder is None:
+ return
+ try:
+ recorder(session_id, cooldown_until, error)
+ except sqlite3.Error as exc:
+ logger.debug("compression failure cooldown persist failed: %s", exc)
+ except Exception as exc:
+ logger.debug("compression failure cooldown persist failed (non-sqlite): %s", exc)
+
+ def _clear_compression_failure_cooldown(self) -> None:
+ self._summary_failure_cooldown_until = 0.0
+ self._last_summary_error = None
+
+ session_db = getattr(self, "_session_db", None)
+ session_id = getattr(self, "_session_id", "")
+ if not session_db or not session_id:
+ return
+
+ clearer = getattr(session_db, "clear_compression_failure_cooldown", None)
+ if clearer is None:
+ return
+ try:
+ clearer(session_id)
+ except sqlite3.Error as exc:
+ logger.debug("compression failure cooldown clear failed: %s", exc)
+ except Exception as exc:
+ logger.debug("compression failure cooldown clear failed (non-sqlite): %s", exc)
+
def update_model(
self,
model: str,
@@ -863,6 +963,8 @@ def __init__(
self.awaiting_real_usage_after_compression = False
self.summary_model = summary_model_override or ""
+ self._session_db: Any = None
+ self._session_id: str = ""
# Stores the previous compaction summary for iterative updates
self._previous_summary: Optional[str] = None
@@ -1448,7 +1550,7 @@ def _fallback_to_main_for_compression(self, e: Exception, reason: str) -> None:
self._last_aux_model_failure_error = _err_text
self._last_aux_model_failure_model = self.summary_model
self.summary_model = "" # empty = use main model
- self._summary_failure_cooldown_until = 0.0 # no cooldown — retry immediately
+ self._clear_compression_failure_cooldown() # no cooldown — retry immediately
def _generate_summary(
self,
@@ -1666,7 +1768,15 @@ def _generate_summary(
# retry (_generate_summary recursion) re-enters harmlessly.
with aux_interrupt_protection():
response = call_llm(**call_kwargs)
- content = response.choices[0].message.content
+ # ``_validate_llm_response`` only guarantees ``choices[0].message``
+ # exists, not that it's an object with ``.content``. Some
+ # OpenAI-compatible proxies / local backends return a dict- or
+ # str-shaped message; coerce defensively instead of crashing.
+ message = response.choices[0].message
+ if isinstance(message, dict):
+ content = message.get("content")
+ else:
+ content = getattr(message, "content", message)
# Handle cases where content is not a string (e.g., dict from llama.cpp)
if not isinstance(content, str):
content = str(content) if content else ""
@@ -1691,7 +1801,7 @@ def _generate_summary(
summary = redact_sensitive_text(content.strip())
# Store for iterative updates on next compaction
self._previous_summary = summary
- self._summary_failure_cooldown_until = 0.0
+ self._clear_compression_failure_cooldown()
self._summary_model_fallen_back = False
self._last_summary_error = None
self._last_summary_auth_failure = False
@@ -1711,7 +1821,10 @@ def _generate_summary(
# a main-model retry before any cooldown. (#11978, #11914)
if isinstance(e, RuntimeError) and "no llm provider configured" in str(e).lower():
# No provider configured — long cooldown, unlikely to self-resolve
- self._summary_failure_cooldown_until = time.monotonic() + _SUMMARY_FAILURE_COOLDOWN_SECONDS
+ self._record_compression_failure_cooldown(
+ _SUMMARY_FAILURE_COOLDOWN_SECONDS,
+ "no auxiliary LLM provider configured",
+ )
self._last_summary_error = "no auxiliary LLM provider configured"
logger.warning("Context compression: no provider available for "
"summary. Middle turns will be dropped without summary "
@@ -1823,10 +1936,10 @@ def _generate_summary(
# streaming premature-close) — shorter cooldown for JSON decode and
# streaming-closed since those conditions can self-resolve quickly.
_transient_cooldown = 30 if (_is_json_decode or _is_streaming_closed) else 60
- self._summary_failure_cooldown_until = time.monotonic() + _transient_cooldown
err_text = str(e).strip() or e.__class__.__name__
if len(err_text) > 220:
err_text = err_text[:217].rstrip() + "..."
+ self._record_compression_failure_cooldown(_transient_cooldown, err_text)
self._last_summary_error = err_text
# A terminal connection/network failure (we reach this branch only
# after any main-model fallback has already been tried or is
@@ -2405,8 +2518,8 @@ def compress(self, messages: List[Dict[str, Any]], current_tokens: int = None, f
# Manual /compress (force=True) bypasses the failure cooldown so the
# user can retry immediately after an auto-compress abort. Without
# this, /compress would silently no-op for 30-60s after a failure.
- if force and self._summary_failure_cooldown_until > 0.0:
- self._summary_failure_cooldown_until = 0.0
+ if force:
+ self._clear_compression_failure_cooldown()
n_messages = len(messages)
# Only need head + 3 tail messages minimum (token budget decides the real tail size)
_min_for_compress = self._protect_head_size(messages) + 3 + 1
diff --git a/agent/context_references.py b/agent/context_references.py
index fad1ff00159b..fe63190e2c0d 100644
--- a/agent/context_references.py
+++ b/agent/context_references.py
@@ -152,13 +152,24 @@ async def preprocess_context_references_async(
blocks: list[str] = []
injected_tokens = 0
- for ref in refs:
- warning, block = await _expand_reference(
- ref,
- cwd_path,
- url_fetcher=url_fetcher,
- allowed_root=allowed_root_path,
+ # Expand all references concurrently. Each _expand_reference is independent
+ # (no shared state during expansion) — a message with several @url: refs
+ # would otherwise pay one full web_extract round-trip per ref in series.
+ # gather preserves positional order, so we reassemble warnings/blocks in the
+ # original ref order exactly as the prior serial loop did; the token-budget
+ # check below is unchanged (it runs once, after all refs are expanded).
+ expanded = await asyncio.gather(
+ *(
+ _expand_reference(
+ ref,
+ cwd_path,
+ url_fetcher=url_fetcher,
+ allowed_root=allowed_root_path,
+ )
+ for ref in refs
)
+ )
+ for warning, block in expanded:
if warning:
warnings.append(warning)
if block:
@@ -328,9 +339,9 @@ async def _fetch_url_content(
async def _default_url_fetcher(url: str) -> str:
from tools.web_tools import web_extract_tool
- raw = await web_extract_tool([url], format="markdown", use_llm_processing=True)
+ raw = await web_extract_tool([url], format="markdown")
payload = json.loads(raw)
- docs = payload.get("data", {}).get("documents", [])
+ docs = payload.get("results", [])
if not docs:
return ""
doc = docs[0]
diff --git a/agent/conversation_compression.py b/agent/conversation_compression.py
index b16765ea9b40..74e9feda2e38 100644
--- a/agent/conversation_compression.py
+++ b/agent/conversation_compression.py
@@ -32,6 +32,7 @@
import os
import tempfile
import uuid
+import threading
from datetime import datetime
from pathlib import Path
from typing import Any, Optional, Tuple
@@ -71,6 +72,85 @@ def _compression_lock_holder(agent: Any) -> str:
)
+class _CompressionLockLeaseRefresher:
+ def __init__(
+ self,
+ db: Any,
+ session_id: str,
+ holder: str,
+ ttl_seconds: float,
+ refresh_interval_seconds: float | None = None,
+ ) -> None:
+ self._db = db
+ self._session_id = session_id
+ self._holder = holder
+ self._ttl_seconds = ttl_seconds
+ if refresh_interval_seconds is None:
+ refresh_interval_seconds = max(1.0, min(60.0, ttl_seconds / 2.0))
+ self._refresh_interval_seconds = max(0.1, float(refresh_interval_seconds))
+ # Tolerate transient refresh failures for at most one lease's worth of
+ # time, so the give-up window is genuinely bounded by the TTL the
+ # acquirer set (a single blip recovers on the next tick; a persistent
+ # failure stops before the lease could outlive its TTL). Floor of 1 so a
+ # degenerate interval >= ttl still tolerates one blip.
+ self._max_consecutive_failures = max(
+ 1, int(self._ttl_seconds / self._refresh_interval_seconds)
+ )
+ self._stop = threading.Event()
+ self._thread = threading.Thread(
+ target=self._run,
+ name="compression-lock-refresh",
+ daemon=True,
+ )
+
+ def start(self) -> "_CompressionLockLeaseRefresher":
+ self._thread.start()
+ return self
+
+ def stop(self) -> None:
+ self._stop.set()
+ # join() may time out while the refresher is mid-UPDATE; that's safe —
+ # it's a daemon thread, and a late refresh on an already-released lock
+ # matches rowcount 0 (a no-op). stop() returning does not guarantee the
+ # thread has fully quiesced, only that we've signalled it and waited
+ # briefly.
+ if self._thread.is_alive() and threading.current_thread() is not self._thread:
+ self._thread.join(timeout=1.0)
+
+ def _run(self) -> None:
+ # A single falsy refresh must NOT permanently kill the lease: a
+ # transient DB blip (write contention escaping _execute_write's retry
+ # budget, a momentary "database is locked") returns False just like a
+ # genuine lost-ownership, but only the latter should stop the loop.
+ # Tolerate consecutive failures for at most one lease's worth of time
+ # (_max_consecutive_failures = ttl / interval), so a one-off blip
+ # recovers on the next tick while the total give-up window stays bounded
+ # by the TTL the acquirer set — the lock can never be held past its TTL
+ # by a stuck refresher.
+ consecutive_failures = 0
+ while not self._stop.wait(self._refresh_interval_seconds):
+ try:
+ refreshed = self._db.refresh_compression_lock(
+ self._session_id,
+ self._holder,
+ ttl_seconds=self._ttl_seconds,
+ )
+ except Exception as exc:
+ logger.debug("compression lock refresh raised: %s", exc)
+ refreshed = False
+ if refreshed:
+ consecutive_failures = 0
+ continue
+ consecutive_failures += 1
+ if consecutive_failures >= self._max_consecutive_failures:
+ logger.debug(
+ "compression lock refresh failed %d times in a row; "
+ "stopping lease refresher for session %s",
+ consecutive_failures, self._session_id,
+ )
+ break
+
+
def check_compression_model_feasibility(agent: Any) -> None:
"""Warn at session start if the auxiliary compression model's context
window is smaller than the main model's compression threshold.
@@ -420,11 +500,17 @@ def compress_context(
# and proceed with compression. Skipping the lock risks a rare
# concurrent-compression session fork; an infinite no-progress loop
# that never compresses at all is strictly worse.
+ try:
+ _lock_ttl = float(getattr(agent, "_compression_lock_ttl_seconds", 300.0) or 300.0)
+ except (TypeError, ValueError):
+ _lock_ttl = 300.0
+ _lock_refresh_interval = getattr(agent, "_compression_lock_refresh_interval", None)
+ _lock_refresher: Optional[_CompressionLockLeaseRefresher] = None
if _lock_db is not None and _lock_sid:
_lock_holder = _compression_lock_holder(agent)
try:
_lock_acquired = _lock_db.try_acquire_compression_lock(
- _lock_sid, _lock_holder
+ _lock_sid, _lock_holder, ttl_seconds=_lock_ttl
)
except Exception as _lock_err:
# Broken/absent lock subsystem (version skew, etc.). Log once
@@ -467,9 +553,19 @@ def compress_context(
if not _existing_sp:
_existing_sp = agent._build_system_prompt(system_message)
return messages, _existing_sp
+ if _lock_holder is not None:
+ _lock_refresher = _CompressionLockLeaseRefresher(
+ _lock_db,
+ _lock_sid,
+ _lock_holder,
+ _lock_ttl,
+ _lock_refresh_interval,
+ ).start()
def _release_lock() -> None:
"""Release the lock keyed on the OLD session_id (before rotation)."""
+ if _lock_refresher is not None:
+ _lock_refresher.stop()
if _lock_db is not None and _lock_sid and _lock_holder:
try:
_lock_db.release_compression_lock(_lock_sid, _lock_holder)
@@ -488,7 +584,11 @@ def _release_lock() -> None:
except TypeError:
# Plugin context engine with strict signature that doesn't accept
# focus_topic / force — fall back to calling without them.
- compressed = agent.context_compressor.compress(messages, current_tokens=approx_tokens)
+ try:
+ compressed = agent.context_compressor.compress(messages, current_tokens=approx_tokens)
+ except BaseException:
+ _release_lock()
+ raise
except BaseException:
# ANY exception during compress() must release the lock so the
# session isn't permanently blocked from future compression.
@@ -501,328 +601,332 @@ def _release_lock() -> None:
# session has logically ended), and let auto-compress callers detect
# the no-op via len(returned) == len(input).
if getattr(agent.context_compressor, "_last_compress_aborted", False):
- _err = getattr(agent.context_compressor, "_last_summary_error", None) or "unknown error"
- if getattr(agent, "_last_compression_summary_warning", None) != _err:
- agent._last_compression_summary_warning = _err
- agent._emit_warning(
- f"⚠ Compression aborted: {_err}. "
- "No messages were dropped — conversation continues unchanged. "
- "Run /compress to retry, or /new to start a fresh session."
- )
- _existing_sp = getattr(agent, "_cached_system_prompt", None)
- if not _existing_sp:
- _existing_sp = agent._build_system_prompt(system_message)
- _release_lock() # compression aborted — no rotation will happen
- return messages, _existing_sp
-
- summary_error = getattr(agent.context_compressor, "_last_summary_error", None)
- if summary_error:
- if getattr(agent, "_last_compression_summary_warning", None) != summary_error:
- agent._last_compression_summary_warning = summary_error
- agent._emit_warning(
- f"⚠ Compression summary failed: {summary_error}. "
- "Inserted a fallback context marker."
- )
- else:
- # No hard failure — but did the configured aux model error out
- # and get recovered by retrying on main? Surface that so users
- # know their auxiliary.compression.model setting is broken even
- # though compression succeeded.
- _aux_fail_model = getattr(agent.context_compressor, "_last_aux_model_failure_model", None)
- _aux_fail_err = getattr(agent.context_compressor, "_last_aux_model_failure_error", None)
- if _aux_fail_model:
- # Dedup on (model, error) so we don't spam on every compaction
- _aux_key = (_aux_fail_model, _aux_fail_err)
- if getattr(agent, "_last_aux_fallback_warning_key", None) != _aux_key:
- agent._last_aux_fallback_warning_key = _aux_key
+ try:
+ _err = getattr(agent.context_compressor, "_last_summary_error", None) or "unknown error"
+ if getattr(agent, "_last_compression_summary_warning", None) != _err:
+ agent._last_compression_summary_warning = _err
agent._emit_warning(
- f"ℹ Configured compression model '{_aux_fail_model}' failed "
- f"({_aux_fail_err or 'unknown error'}). Recovered using main model — "
- "check auxiliary.compression.model in config.yaml."
+ f"⚠ Compression aborted: {_err}. "
+ "No messages were dropped — conversation continues unchanged. "
+ "Run /compress to retry, or /new to start a fresh session."
)
+ _existing_sp = getattr(agent, "_cached_system_prompt", None)
+ if not _existing_sp:
+ _existing_sp = agent._build_system_prompt(system_message)
+ return messages, _existing_sp
+ finally:
+ _release_lock()
- todo_snapshot = agent._todo_store.format_for_injection()
- if todo_snapshot:
- compressed.append({"role": "user", "content": todo_snapshot})
-
- agent._invalidate_system_prompt()
- new_system_prompt = agent._build_system_prompt(system_message)
- agent._cached_system_prompt = new_system_prompt
+ try:
+ summary_error = getattr(agent.context_compressor, "_last_summary_error", None)
+ if summary_error:
+ if getattr(agent, "_last_compression_summary_warning", None) != summary_error:
+ agent._last_compression_summary_warning = summary_error
+ agent._emit_warning(
+ f"⚠ Compression summary failed: {summary_error}. "
+ "Inserted a fallback context marker."
+ )
+ else:
+ # No hard failure — but did the configured aux model error out
+ # and get recovered by retrying on main? Surface that so users
+ # know their auxiliary.compression.model setting is broken even
+ # though compression succeeded.
+ _aux_fail_model = getattr(agent.context_compressor, "_last_aux_model_failure_model", None)
+ _aux_fail_err = getattr(agent.context_compressor, "_last_aux_model_failure_error", None)
+ if _aux_fail_model:
+ # Dedup on (model, error) so we don't spam on every compaction
+ _aux_key = (_aux_fail_model, _aux_fail_err)
+ if getattr(agent, "_last_aux_fallback_warning_key", None) != _aux_key:
+ agent._last_aux_fallback_warning_key = _aux_key
+ agent._emit_warning(
+ f"ℹ Configured compression model '{_aux_fail_model}' failed "
+ f"({_aux_fail_err or 'unknown error'}). Recovered using main model — "
+ "check auxiliary.compression.model in config.yaml."
+ )
- if agent._session_db:
- try:
- # Trigger memory extraction on the current session before the
- # transcript is rewritten (runs in BOTH modes — the logical
- # conversation's pre-compaction turns are about to be summarized
- # away regardless of whether the id rotates).
- agent.commit_memory_session(messages)
-
- if in_place:
- # ── In-place compaction: keep the same session_id ──────────
- # No end_session, no new row, no parent_session_id, no title
- # renumber, no contextvar/env/logging re-sync. The session's
- # id, title, cwd, /goal, and gateway routing all stay put.
- #
- # Durable, NON-DESTRUCTIVE replace: soft-archive the
- # pre-compaction turns (active=0, kept on disk + FTS-searchable +
- # recoverable) and insert `compressed` as the new live (active=1)
- # set, atomically. `compressed` already carries the surviving
- # tail (current-turn messages the compressor kept via
- # protect_last_n), so we DON'T pre-flush here — a flush would
- # INSERT current-turn rows that archive_and_compact would then
- # archive alongside the rest (harmless but wasted writes). The
- # live-context load filters active=1, so a resume reloads ONLY
- # the compacted set; the original turns remain under the SAME id
- # for search/recovery (Teknium review — keep one durable id
- # WITHOUT destroying history, unlike a hard replace_messages).
- # See #38763.
- agent._session_db.archive_and_compact(agent.session_id, compressed)
- # Reset the flush identity set so the next turn's appends are
- # diffed against the COMPACTED transcript: the compacted dicts
- # are passed as conversation_history next turn and skipped by
- # identity, so only genuinely new turn messages get appended
- # (no dup of the summary, no resurrection of dropped turns).
- agent._flushed_db_message_ids = set()
- # Rotation-independent signal: the conversation was compacted in
- # place (id unchanged). The gateway reads this (NOT an id-change
- # diff) to re-baseline transcript handling.
- compacted_in_place = True
- else:
- # ── Rotation (legacy): end this session, fork a continuation ─
- # Flush any un-persisted current-turn messages to the OLD
- # session before ending it, so they survive in the preserved
- # parent transcript (#47202). (In-place skips this — see above.)
- try:
- agent._flush_messages_to_session_db(messages)
- except Exception:
- pass # best-effort — don't block compression on a flush error
- # Propagate title to the new session with auto-numbering
- old_title = agent._session_db.get_session_title(agent.session_id)
- agent._session_db.end_session(agent.session_id, "compression")
- old_session_id = agent.session_id
- agent.session_id = f"{datetime.now().strftime('%Y%m%d_%H%M%S')}_{uuid.uuid4().hex[:6]}"
- # Ordering contract: the agent thread updates the contextvar here;
- # the gateway propagates to SessionEntry after run_in_executor returns.
- try:
- from gateway.session_context import set_current_session_id
+ todo_snapshot = agent._todo_store.format_for_injection()
+ if todo_snapshot:
+ compressed.append({"role": "user", "content": todo_snapshot})
- set_current_session_id(agent.session_id)
- except Exception:
- os.environ["HERMES_SESSION_ID"] = agent.session_id
- # The gateway/tools session context (ContextVar + env) and the
- # logging session context are SEPARATE mechanisms. The call above
- # moves the former; the ``[session_id]`` tag on log lines comes
- # from ``hermes_logging._session_context`` (set once per turn in
- # conversation_loop.py). Without this, post-rotation log lines in
- # the same turn keep the STALE old id while the message/DB/gateway
- # state carry the new one — breaking log correlation exactly at the
- # compaction boundary (see #34089). Guarded separately so a logging
- # failure can never regress the routing update above.
- try:
- from hermes_logging import set_session_context
+ agent._invalidate_system_prompt()
+ new_system_prompt = agent._build_system_prompt(system_message)
+ agent._cached_system_prompt = new_system_prompt
- set_session_context(agent.session_id)
- except Exception:
- pass
- agent._session_db_created = False
- try:
- agent._session_db.create_session(
- session_id=agent.session_id,
- source=agent.platform or os.environ.get("HERMES_SESSION_SOURCE", "cli"),
- model=agent.model,
- model_config=agent._session_init_model_config,
- parent_session_id=old_session_id,
- )
- except Exception as _cs_err:
- # The child row could not be created (e.g. FK constraint,
- # contended write). Previously the outer handler simply
- # warned and let the agent continue on the NEW id — which
- # has no row in state.db, producing an orphan: the parent
- # is ended, the child is never indexed, and every
- # subsequent message is attributed to a session that
- # doesn't exist (#33906/#33907). Roll the live id back to
- # the parent so the conversation stays attached to a real,
- # indexed session instead of a phantom.
- logger.warning(
- "Compression child session create failed (%s) — "
- "rolling back to parent session %s to avoid an orphan.",
- _cs_err, old_session_id,
- )
- agent.session_id = old_session_id
+ if agent._session_db:
+ try:
+ # Trigger memory extraction on the current session before the
+ # transcript is rewritten (runs in BOTH modes — the logical
+ # conversation's pre-compaction turns are about to be summarized
+ # away regardless of whether the id rotates).
+ agent.commit_memory_session(messages)
+
+ if in_place:
+ # ── In-place compaction: keep the same session_id ──────────
+ # No end_session, no new row, no parent_session_id, no title
+ # renumber, no contextvar/env/logging re-sync. The session's
+ # id, title, cwd, /goal, and gateway routing all stay put.
+ #
+ # Durable, NON-DESTRUCTIVE replace: soft-archive the
+ # pre-compaction turns (active=0, kept on disk + FTS-searchable +
+ # recoverable) and insert `compressed` as the new live (active=1)
+ # set, atomically. `compressed` already carries the surviving
+ # tail (current-turn messages the compressor kept via
+ # protect_last_n), so we DON'T pre-flush here — a flush would
+ # INSERT current-turn rows that archive_and_compact would then
+ # archive alongside the rest (harmless but wasted writes). The
+ # live-context load filters active=1, so a resume reloads ONLY
+ # the compacted set; the original turns remain under the SAME id
+ # for search/recovery (Teknium review — keep one durable id
+ # WITHOUT destroying history, unlike a hard replace_messages).
+ # See #38763.
+ agent._session_db.archive_and_compact(agent.session_id, compressed)
+ # Reset the flush identity set so the next turn's appends are
+ # diffed against the COMPACTED transcript: the compacted dicts
+ # are passed as conversation_history next turn and skipped by
+ # identity, so only genuinely new turn messages get appended
+ # (no dup of the summary, no resurrection of dropped turns).
+ agent._flushed_db_message_ids = set()
+ # Rotation-independent signal: the conversation was compacted in
+ # place (id unchanged). The gateway reads this (NOT an id-change
+ # diff) to re-baseline transcript handling.
+ compacted_in_place = True
+ else:
+ # ── Rotation (legacy): end this session, fork a continuation ─
+ # Flush any un-persisted current-turn messages to the OLD
+ # session before ending it, so they survive in the preserved
+ # parent transcript (#47202). (In-place skips this — see above.)
+ try:
+ agent._flush_messages_to_session_db(messages)
+ except Exception:
+ pass # best-effort — don't block compression on a flush error
+ # Propagate title to the new session with auto-numbering
+ old_title = agent._session_db.get_session_title(agent.session_id)
+ agent._session_db.end_session(agent.session_id, "compression")
+ old_session_id = agent.session_id
+ agent.session_id = f"{datetime.now().strftime('%Y%m%d_%H%M%S')}_{uuid.uuid4().hex[:6]}"
+ # Ordering contract: the agent thread updates the contextvar here;
+ # the gateway propagates to SessionEntry after run_in_executor returns.
try:
from gateway.session_context import set_current_session_id
+
set_current_session_id(agent.session_id)
except Exception:
os.environ["HERMES_SESSION_ID"] = agent.session_id
+ # The gateway/tools session context (ContextVar + env) and the
+ # logging session context are SEPARATE mechanisms. The call above
+ # moves the former; the ``[session_id]`` tag on log lines comes
+ # from ``hermes_logging._session_context`` (set once per turn in
+ # conversation_loop.py). Without this, post-rotation log lines in
+ # the same turn keep the STALE old id while the message/DB/gateway
+ # state carry the new one — breaking log correlation exactly at the
+ # compaction boundary (see #34089). Guarded separately so a logging
+ # failure can never regress the routing update above.
try:
from hermes_logging import set_session_context
+
set_session_context(agent.session_id)
except Exception:
pass
- # Re-open the parent: it was ended above, but we're
- # continuing on it, so it must not stay closed.
+ agent._session_db_created = False
try:
- agent._session_db.reopen_session(old_session_id)
- except Exception:
- pass
- old_session_id = None # no rotation happened
- # The parent row already exists in state.db, so mark the
- # session as created — _ensure_db_session would otherwise
- # retry a (harmless INSERT OR IGNORE) create next turn.
+ agent._session_db.create_session(
+ session_id=agent.session_id,
+ source=agent.platform or os.environ.get("HERMES_SESSION_SOURCE", "cli"),
+ model=agent.model,
+ model_config=agent._session_init_model_config,
+ parent_session_id=old_session_id,
+ )
+ except Exception as _cs_err:
+ # The child row could not be created (e.g. FK constraint,
+ # contended write). Previously the outer handler simply
+ # warned and let the agent continue on the NEW id — which
+ # has no row in state.db, producing an orphan: the parent
+ # is ended, the child is never indexed, and every
+ # subsequent message is attributed to a session that
+ # doesn't exist (#33906/#33907). Roll the live id back to
+ # the parent so the conversation stays attached to a real,
+ # indexed session instead of a phantom.
+ logger.warning(
+ "Compression child session create failed (%s) — "
+ "rolling back to parent session %s to avoid an orphan.",
+ _cs_err, old_session_id,
+ )
+ agent.session_id = old_session_id
+ try:
+ from gateway.session_context import set_current_session_id
+ set_current_session_id(agent.session_id)
+ except Exception:
+ os.environ["HERMES_SESSION_ID"] = agent.session_id
+ try:
+ from hermes_logging import set_session_context
+ set_session_context(agent.session_id)
+ except Exception:
+ pass
+ # Re-open the parent: it was ended above, but we're
+ # continuing on it, so it must not stay closed.
+ try:
+ agent._session_db.reopen_session(old_session_id)
+ except Exception:
+ pass
+ old_session_id = None # no rotation happened
+ # The parent row already exists in state.db, so mark the
+ # session as created — _ensure_db_session would otherwise
+ # retry a (harmless INSERT OR IGNORE) create next turn.
+ agent._session_db_created = True
+ raise
agent._session_db_created = True
- raise
- agent._session_db_created = True
- # Carry a persistent /goal onto the continuation session.
- # Compression mints a fresh child id; load_goal does a flat
- # per-session lookup with no parent walk, so without this an
- # active goal silently dies at the boundary (#33618).
- try:
- from hermes_cli.goals import migrate_goal_to_session
- migrate_goal_to_session(old_session_id, agent.session_id, reason="compression")
- except Exception as _goal_err:
- logger.debug("Could not migrate goal on compression: %s", _goal_err)
- # Auto-number the title for the continuation session
- if old_title:
+ # Carry a persistent /goal onto the continuation session.
+ # Compression mints a fresh child id; load_goal does a flat
+ # per-session lookup with no parent walk, so without this an
+ # active goal silently dies at the boundary (#33618).
try:
- new_title = agent._session_db.get_next_title_in_lineage(old_title)
- agent._session_db.set_session_title(agent.session_id, new_title)
- except (ValueError, Exception) as e:
- logger.debug("Could not propagate title on compression: %s", e)
-
- # Shared post-write steps (both modes target agent.session_id, which
- # in-place keeps and rotation has already reassigned to the new id):
- # refresh the stored system prompt and reset the flush cursor so the
- # next turn re-bases its append diff.
- agent._session_db.update_system_prompt(agent.session_id, new_system_prompt)
- agent._last_flushed_db_idx = 0
- except Exception as e:
- # If the rotation rolled back to the parent (orphan-avoidance
- # above), agent.session_id is the still-indexed parent and
- # old_session_id was cleared — so this is recovery, not an
- # un-indexed orphan. Otherwise an earlier step failed before the
- # child was created and the warning's original meaning holds.
- if locals().get("old_session_id") is None and not in_place:
- logger.warning(
- "Compression rotation aborted and rolled back to the "
- "parent session (%s): %s", agent.session_id or "?", e,
+ from hermes_cli.goals import migrate_goal_to_session
+ migrate_goal_to_session(old_session_id, agent.session_id, reason="compression")
+ except Exception as _goal_err:
+ logger.debug("Could not migrate goal on compression: %s", _goal_err)
+ # Auto-number the title for the continuation session
+ if old_title:
+ try:
+ new_title = agent._session_db.get_next_title_in_lineage(old_title)
+ agent._session_db.set_session_title(agent.session_id, new_title)
+ except (ValueError, Exception) as e:
+ logger.debug("Could not propagate title on compression: %s", e)
+
+ # Shared post-write steps (both modes target agent.session_id, which
+ # in-place keeps and rotation has already reassigned to the new id):
+ # refresh the stored system prompt and reset the flush cursor so the
+ # next turn re-bases its append diff.
+ agent._session_db.update_system_prompt(agent.session_id, new_system_prompt)
+ agent._last_flushed_db_idx = 0
+ except Exception as e:
+ # If the rotation rolled back to the parent (orphan-avoidance
+ # above), agent.session_id is the still-indexed parent and
+ # old_session_id was cleared — so this is recovery, not an
+ # un-indexed orphan. Otherwise an earlier step failed before the
+ # child was created and the warning's original meaning holds.
+ if locals().get("old_session_id") is None and not in_place:
+ logger.warning(
+ "Compression rotation aborted and rolled back to the "
+ "parent session (%s): %s", agent.session_id or "?", e,
+ )
+ else:
+ logger.warning("Session DB compression split failed — new session will NOT be indexed: %s", e)
+
+ # Compaction-boundary bookkeeping, computed once. `old_session_id` is only
+ # bound in the rotation branch; in-place leaves it unset. `_boundary_parent`
+ # is the id the boundary notifications attribute the prior state to: the old
+ # id on rotation, the (unchanged) current id in-place.
+ _old_sid = locals().get("old_session_id")
+ _is_boundary = bool(_old_sid) or in_place
+ _boundary_parent = _old_sid or agent.session_id or ""
+
+ # Notify the context engine that a compaction boundary occurred. Plugin
+ # engines (e.g. hermes-lcm) use boundary_reason="compression" to preserve
+ # DAG lineage / checkpoint per-session state across the boundary instead of
+ # re-initializing fresh. See hermes-lcm#68. Built-in ContextCompressor
+ # ignores kwargs. Fires in BOTH modes: rotation passes old→new ids; in-place
+ # passes the SAME id (the boundary is real even though the id didn't move).
+ try:
+ if _is_boundary and hasattr(agent.context_compressor, "on_session_start"):
+ agent.context_compressor.on_session_start(
+ agent.session_id or "",
+ boundary_reason="compression",
+ old_session_id=_boundary_parent,
+ platform=getattr(agent, "platform", None) or "cli",
+ conversation_id=getattr(agent, "_gateway_session_key", None),
)
- else:
- logger.warning("Session DB compression split failed — new session will NOT be indexed: %s", e)
-
- # Compaction-boundary bookkeeping, computed once. `old_session_id` is only
- # bound in the rotation branch; in-place leaves it unset. `_boundary_parent`
- # is the id the boundary notifications attribute the prior state to: the old
- # id on rotation, the (unchanged) current id in-place.
- _old_sid = locals().get("old_session_id")
- _is_boundary = bool(_old_sid) or in_place
- _boundary_parent = _old_sid or agent.session_id or ""
-
- # Notify the context engine that a compaction boundary occurred. Plugin
- # engines (e.g. hermes-lcm) use boundary_reason="compression" to preserve
- # DAG lineage / checkpoint per-session state across the boundary instead of
- # re-initializing fresh. See hermes-lcm#68. Built-in ContextCompressor
- # ignores kwargs. Fires in BOTH modes: rotation passes old→new ids; in-place
- # passes the SAME id (the boundary is real even though the id didn't move).
- try:
- if _is_boundary and hasattr(agent.context_compressor, "on_session_start"):
- agent.context_compressor.on_session_start(
- agent.session_id or "",
- boundary_reason="compression",
- old_session_id=_boundary_parent,
- platform=getattr(agent, "platform", None) or "cli",
- conversation_id=getattr(agent, "_gateway_session_key", None),
- )
- except Exception as _ce_err:
- logger.debug("context engine on_session_start (compression): %s", _ce_err)
-
- # Notify memory providers of the compaction boundary so provider-cached
- # per-session state (Hindsight's _document_id, accumulated turn buffers,
- # counters) refreshes. reset=False because the logical conversation
- # continues. See #6672. Fires in BOTH modes: in-place uses the same id as
- # parent (the conversation didn't fork, but the buffer must still be told
- # the transcript was compacted so it doesn't double-count dropped turns).
- try:
- if _is_boundary and agent._memory_manager:
- agent._memory_manager.on_session_switch(
- agent.session_id or "",
- parent_session_id=_boundary_parent,
- reset=False,
- reason="compression",
+ except Exception as _ce_err:
+ logger.debug("context engine on_session_start (compression): %s", _ce_err)
+
+ # Notify memory providers of the compaction boundary so provider-cached
+ # per-session state (Hindsight's _document_id, accumulated turn buffers,
+ # counters) refreshes. reset=False because the logical conversation
+ # continues. See #6672. Fires in BOTH modes: in-place uses the same id as
+ # parent (the conversation didn't fork, but the buffer must still be told
+ # the transcript was compacted so it doesn't double-count dropped turns).
+ try:
+ if _is_boundary and agent._memory_manager:
+ agent._memory_manager.on_session_switch(
+ agent.session_id or "",
+ parent_session_id=_boundary_parent,
+ reset=False,
+ reason="compression",
+ )
+ except Exception as _me_err:
+ logger.debug("memory manager on_session_switch (compression): %s", _me_err)
+
+ # Warn on repeated compressions (quality degrades with each pass).
+ # Route through _emit_status (like the other compression warnings above)
+ # so the warning reaches the TUI / Telegram / Discord via status_callback,
+ # not just CLI stdout. _emit_status still _vprints for the CLI, and
+ # storing it on _compression_warning lets replay_compression_warning
+ # re-deliver it once a late-bound gateway status_callback is wired (#36908).
+ _cc = agent.context_compressor.compression_count
+ if _cc >= 2:
+ _cc_msg = (
+ f"{agent.log_prefix}⚠️ Session compressed {_cc} times — "
+ f"accuracy may degrade. Consider /new to start fresh."
)
- except Exception as _me_err:
- logger.debug("memory manager on_session_switch (compression): %s", _me_err)
-
- # Warn on repeated compressions (quality degrades with each pass).
- # Route through _emit_status (like the other compression warnings above)
- # so the warning reaches the TUI / Telegram / Discord via status_callback,
- # not just CLI stdout. _emit_status still _vprints for the CLI, and
- # storing it on _compression_warning lets replay_compression_warning
- # re-deliver it once a late-bound gateway status_callback is wired (#36908).
- _cc = agent.context_compressor.compression_count
- if _cc >= 2:
- _cc_msg = (
- f"{agent.log_prefix}⚠️ Session compressed {_cc} times — "
- f"accuracy may degrade. Consider /new to start fresh."
+ agent._compression_warning = _cc_msg
+ agent._emit_status(_cc_msg)
+
+ # Emit session:compress event so hooks (e.g. MemPalace sync) can ingest
+ # the completed old session before its details are lost. In in-place mode
+ # there is no old id (same session); ``in_place=True`` tells hooks the
+ # transcript was compacted on the same id rather than rotated.
+ if getattr(agent, "event_callback", None):
+ try:
+ agent.event_callback("session:compress", {
+ "platform": agent.platform or "",
+ "session_id": agent.session_id,
+ "old_session_id": _old_sid or "",
+ "in_place": in_place,
+ "compression_count": agent.context_compressor.compression_count,
+ })
+ except Exception as e:
+ logger.debug("event_callback error on session:compress: %s", e)
+
+ # Surface the compaction mode to the caller (run_conversation / gateway)
+ # via a rotation-independent flag. The gateway uses this — NOT an
+ # id-change diff — to re-baseline transcript handling (history_offset=0 +
+ # rewrite on the same id) when compaction happened in place. See #38763.
+ agent._last_compaction_in_place = compacted_in_place
+
+ # Keep the post-compression rough estimate for diagnostics, but do not
+ # treat it as provider-reported prompt usage. Schema-heavy rough estimates
+ # can remain above threshold even after the next real API request fits.
+ _compressed_est = estimate_request_tokens_rough(
+ compressed,
+ system_prompt=new_system_prompt or "",
+ tools=agent.tools or None,
)
- agent._compression_warning = _cc_msg
- agent._emit_status(_cc_msg)
-
- # Emit session:compress event so hooks (e.g. MemPalace sync) can ingest
- # the completed old session before its details are lost. In in-place mode
- # there is no old id (same session); ``in_place=True`` tells hooks the
- # transcript was compacted on the same id rather than rotated.
- if getattr(agent, "event_callback", None):
+ agent.context_compressor.last_compression_rough_tokens = _compressed_est
+ agent.context_compressor.last_prompt_tokens = -1
+ agent.context_compressor.last_completion_tokens = 0
+ agent.context_compressor.awaiting_real_usage_after_compression = True
+
+ # Clear the file-read dedup cache. After compression the original
+ # read content is summarised away — if the model re-reads the same
+ # file it needs the full content, not a "file unchanged" stub.
try:
- agent.event_callback("session:compress", {
- "platform": agent.platform or "",
- "session_id": agent.session_id,
- "old_session_id": _old_sid or "",
- "in_place": in_place,
- "compression_count": agent.context_compressor.compression_count,
- })
- except Exception as e:
- logger.debug("event_callback error on session:compress: %s", e)
-
- # Surface the compaction mode to the caller (run_conversation / gateway)
- # via a rotation-independent flag. The gateway uses this — NOT an
- # id-change diff — to re-baseline transcript handling (history_offset=0 +
- # rewrite on the same id) when compaction happened in place. See #38763.
- agent._last_compaction_in_place = compacted_in_place
-
- # Keep the post-compression rough estimate for diagnostics, but do not
- # treat it as provider-reported prompt usage. Schema-heavy rough estimates
- # can remain above threshold even after the next real API request fits.
- _compressed_est = estimate_request_tokens_rough(
- compressed,
- system_prompt=new_system_prompt or "",
- tools=agent.tools or None,
- )
- agent.context_compressor.last_compression_rough_tokens = _compressed_est
- agent.context_compressor.last_prompt_tokens = -1
- agent.context_compressor.last_completion_tokens = 0
- agent.context_compressor.awaiting_real_usage_after_compression = True
-
- # Clear the file-read dedup cache. After compression the original
- # read content is summarised away — if the model re-reads the same
- # file it needs the full content, not a "file unchanged" stub.
- try:
- from tools.file_tools import reset_file_dedup
- reset_file_dedup(task_id)
- except Exception:
- pass
+ from tools.file_tools import reset_file_dedup
+ reset_file_dedup(task_id)
+ except Exception:
+ pass
- logger.info(
- "context compression done: session=%s messages=%d->%d rough_tokens=~%s awaiting_real_usage=true",
- agent.session_id or "none", _pre_msg_count, len(compressed),
- f"{_compressed_est:,}",
- )
- # Release the lock on the OLD session_id only AFTER rotation completed
- # and all post-rotation bookkeeping (memory manager, context engine,
- # file dedup) ran. A concurrent path that wakes up the moment we
- # release will see the NEW session_id in state.db / SessionEntry and
- # acquire on that — no race against our just-finished work.
- _release_lock()
- return compressed, new_system_prompt
+ logger.info(
+ "context compression done: session=%s messages=%d->%d rough_tokens=~%s awaiting_real_usage=true",
+ agent.session_id or "none", _pre_msg_count, len(compressed),
+ f"{_compressed_est:,}",
+ )
+ return compressed, new_system_prompt
+ finally:
+ # Release the lock on the OLD session_id only AFTER rotation completed
+ # and all post-rotation bookkeeping (memory manager, context engine,
+ # file dedup) ran. A concurrent path that wakes up the moment we
+ # release will see the NEW session_id in state.db / SessionEntry and
+ # acquire on that — no race against our just-finished work.
+ _release_lock()
def try_shrink_image_parts_in_messages(
diff --git a/agent/conversation_loop.py b/agent/conversation_loop.py
index 10825cfd683e..7a5919807af1 100644
--- a/agent/conversation_loop.py
+++ b/agent/conversation_loop.py
@@ -52,6 +52,7 @@
estimate_messages_tokens_rough,
estimate_request_tokens_rough,
get_context_length_from_provider_error,
+ is_output_cap_error,
parse_available_output_tokens_from_error,
save_context_length,
)
@@ -1167,11 +1168,22 @@ def _stop_spinner():
# stream. Mirror the ACP exclusion used for Responses
# API upgrade (lines ~1083-1085).
elif (
- agent.provider in {"copilot-acp", "moa"}
+ agent.provider in {"copilot-acp"}
or str(agent.base_url or "").lower().startswith("acp://copilot")
or str(agent.base_url or "").lower().startswith("acp+tcp://")
):
_use_streaming = False
+ # MoA streams only when a display/TTS consumer is present to
+ # receive the deltas. MoAChatCompletions.create() honors
+ # stream=True (runs the references, then returns the aggregator's
+ # raw token stream) and is reached here because, for provider
+ # "moa", _create_request_openai_client returns the MoA facade
+ # itself. Without consumers (quiet mode, subagents, health-check
+ # probes) we keep the complete-response path: the facade returns a
+ # whole response when stream is not requested, preserving the
+ # prior behavior for those callers.
+ elif agent.provider == "moa" and not agent._has_stream_consumers():
+ _use_streaming = False
elif not agent._has_stream_consumers():
# No display/TTS consumer. Still prefer streaming for
# health checking, but skip for Mock clients in tests
@@ -2919,6 +2931,7 @@ def _perform_api_call(next_api_kwargs):
is_rate_limited = classified.reason in {
FailoverReason.rate_limit,
FailoverReason.billing,
+ FailoverReason.upstream_rate_limit,
}
_is_transport_failure = classified.reason in {
FailoverReason.timeout,
@@ -2933,13 +2946,30 @@ def _perform_api_call(next_api_kwargs):
# still recover. See _pool_may_recover_from_rate_limit
# for the single-credential-pool and CloudCode-quota
# exceptions. Fixes #11314 and #13636.
- pool_may_recover = _ra()._pool_may_recover_from_rate_limit(
- agent._credential_pool,
- provider=agent.provider,
- base_url=getattr(agent, "base_url", None),
+ #
+ # Exception: an upstream-aggregator 429 — the credential
+ # pool can't help when the *upstream* model (DeepSeek,
+ # etc.) is throttling OpenRouter, so always fall back to a
+ # different model regardless of pool state.
+ _is_upstream = classified.reason == FailoverReason.upstream_rate_limit
+ pool_may_recover = (
+ False if _is_upstream
+ else _ra()._pool_may_recover_from_rate_limit(
+ agent._credential_pool,
+ provider=agent.provider,
+ base_url=getattr(agent, "base_url", None),
+ )
)
if not pool_may_recover:
- if classified.reason == FailoverReason.billing:
+ if _is_upstream:
+ _upstream_name = (classified.error_context or {}).get(
+ "upstream_provider", "aggregator"
+ )
+ agent._buffer_status(
+ f"⚠️ Upstream {_upstream_name} rate-limited — "
+ "switching to fallback model..."
+ )
+ elif classified.reason == FailoverReason.billing:
agent._buffer_status(
"⚠️ Billing or credits exhausted — switching to fallback provider..."
)
@@ -3213,6 +3243,45 @@ def _perform_api_call(next_api_kwargs):
_retry.restart_with_compressed_messages = True
break
+ # The error is output-cap-shaped (about max_tokens being
+ # too large) but the provider's wording didn't let us parse
+ # the available output budget. Compression CANNOT help here
+ # — the input already fits; the call fails deterministically
+ # on the oversized max_tokens. Routing it into compression
+ # re-sends the same max_tokens, gets the identical 400, and
+ # death-loops until "cannot compress further" (#55546).
+ # Fail fast with an actionable message instead of looping.
+ if is_output_cap_error(error_msg):
+ agent._flush_status_buffer()
+ agent._vprint(
+ f"{agent.log_prefix}❌ The provider rejected the request because "
+ f"max_tokens exceeds its output cap for this model.",
+ force=True,
+ )
+ agent._vprint(
+ f"{agent.log_prefix} 💡 Lower model.max_tokens in your config.yaml to "
+ f"at or below the model's max-output limit. "
+ f"(This is an output-cap error, not a context overflow — "
+ f"compression cannot fix it.)",
+ force=True,
+ )
+ logger.error(
+ f"{agent.log_prefix}Output-cap error not routed into compression "
+ f"(max_tokens over provider cap): {error_msg[:200]}"
+ )
+ agent._persist_session(messages, conversation_history)
+ return {
+ "messages": messages,
+ "completed": False,
+ "api_calls": api_call_count,
+ "error": (
+ "max_tokens exceeds the provider's output cap for this model. "
+ "Lower model.max_tokens in config.yaml."
+ ),
+ "partial": True,
+ "failed": True,
+ }
+
# Error is about the INPUT being too large. Only reduce
# context_length when the provider explicitly reports the
# real lower limit. If the provider only says "input
@@ -4810,6 +4879,55 @@ def _perform_api_call(next_api_kwargs):
agent._verification_stop_nudges)
continue
+ # User verification-loop gate: when the agent edited code this
+ # turn, let a registered `pre_verify` hook (plugin/shell) keep it
+ # going one more turn. The shipped guidance is folded into the
+ # evidence-based verify-on-stop nudge above, so this path has no
+ # default continuation cost.
+ _verify_nudge2 = None
+ _edited = sorted(getattr(agent, "_turn_file_mutation_paths", set()) or [])
+ _attempt = getattr(agent, "_pre_verify_nudges", 0)
+ try:
+ from agent.verify_hooks import max_verify_nudges
+ from hermes_cli.plugins import get_pre_verify_continue_message, has_hook
+
+ if _edited and has_hook("pre_verify") and _attempt < max_verify_nudges():
+ # Posture is fixed for the session — resolve once + cache.
+ coding = getattr(agent, "_resolved_is_coding", None)
+ if coding is None:
+ from agent.coding_context import is_coding_context
+ coding = bool(is_coding_context(platform=getattr(agent, "platform", "") or ""))
+ agent._resolved_is_coding = coding
+ _verify_nudge2 = get_pre_verify_continue_message(
+ session_id=getattr(agent, "session_id", None) or "",
+ platform=getattr(agent, "platform", "") or "",
+ model=getattr(agent, "model", "") or "",
+ coding=coding,
+ attempt=_attempt,
+ final_response=final_response,
+ changed_paths=_edited,
+ )
+ except Exception:
+ logger.debug("pre_verify hook check failed", exc_info=True)
+ _verify_nudge2 = None
+
+ if _verify_nudge2:
+ agent._pre_verify_nudges = _attempt + 1
+ final_msg["finish_reason"] = "verify_hook_continue"
+ # Same alternation contract as verify-on-stop: keep the
+ # attempted answer in history, follow it with a synthetic
+ # user nudge, and don't surface the premature answer.
+ messages.append(final_msg)
+ messages.append({
+ "role": "user",
+ "content": _verify_nudge2,
+ "_pre_verify_synthetic": True,
+ })
+ agent._session_messages = messages
+ logger.debug("pre_verify nudge issued (attempt %d)",
+ agent._pre_verify_nudges)
+ continue
+
messages.append(final_msg)
_turn_exit_reason = f"text_response(finish_reason={finish_reason})"
diff --git a/agent/credential_pool.py b/agent/credential_pool.py
index d8ca2b1720ea..8d10bbb1cbf7 100644
--- a/agent/credential_pool.py
+++ b/agent/credential_pool.py
@@ -616,17 +616,32 @@ def _sync_anthropic_entry_from_credentials_file(self, entry: PooledCredential) -
file_refresh = creds.get("refreshToken", "")
file_access = creds.get("accessToken", "")
file_expires = creds.get("expiresAt", 0)
- # If the credentials file has a different token pair, sync it
- if file_refresh and file_refresh != entry.refresh_token:
- logger.debug("Pool entry %s: syncing tokens from credentials file (refresh token changed)", entry.id)
+ # Sync when either token changed. Access tokens can be re-issued
+ # without a new refresh token (silent re-issue path), so checking
+ # only refresh_token misses that case and leaves a stale
+ # access_token in the pool → 401 on every request until the pool
+ # entry's exhausted TTL expires.
+ entry_access = entry.access_token or ""
+ entry_refresh = entry.refresh_token or ""
+ if (file_access or file_refresh) and (
+ (file_access and file_access != entry_access)
+ or (file_refresh and file_refresh != entry_refresh)
+ ):
+ logger.debug(
+ "Pool entry %s: syncing tokens from credentials file (tokens changed)",
+ entry.id,
+ )
updated = replace(
entry,
- access_token=file_access,
- refresh_token=file_refresh,
- expires_at_ms=file_expires,
+ access_token=file_access or entry.access_token,
+ refresh_token=file_refresh or entry.refresh_token,
+ expires_at_ms=file_expires or entry.expires_at_ms,
last_status=None,
last_status_at=None,
last_error_code=None,
+ last_error_reason=None,
+ last_error_message=None,
+ last_error_reset_at=None,
)
self._replace_entry(entry, updated)
self._persist()
@@ -1884,11 +1899,16 @@ def _env_val(key: str) -> str:
from hermes_cli.copilot_auth import resolve_copilot_token, get_copilot_api_token
token, source = resolve_copilot_token()
if token:
- api_token = get_copilot_api_token(token)
+ api_token, enterprise_base_url = get_copilot_api_token(token)
source_name = "gh_cli" if "gh" in source.lower() else f"env:{source}"
if not _is_suppressed(provider, source_name):
active_sources.add(source_name)
pconfig = PROVIDER_REGISTRY.get(provider)
+ # Use enterprise base URL from token exchange if available,
+ # otherwise fall back to the provider's default.
+ effective_base_url = enterprise_base_url or (
+ pconfig.inference_base_url if pconfig else ""
+ )
changed |= _upsert_entry(
entries,
provider,
@@ -1897,7 +1917,7 @@ def _env_val(key: str) -> str:
"source": source_name,
"auth_type": AUTH_TYPE_API_KEY,
"access_token": api_token,
- "base_url": pconfig.inference_base_url if pconfig else "",
+ "base_url": effective_base_url,
"label": source,
},
)
diff --git a/agent/display.py b/agent/display.py
index 861d84bc4105..060ac1266fa0 100644
--- a/agent/display.py
+++ b/agent/display.py
@@ -537,6 +537,122 @@ def build_tool_preview(tool_name: str, args: dict, max_len: int | None = None) -
return preview
+# =========================================================================
+# Friendly tool labels (human-phrased verbs for built-in tools)
+#
+# Turns "web_search " into "Searching the web for " — the
+# ChatGPT-style "Searching…/Reading…" surface. Curated and built-in only:
+# we know each core tool's semantics, so the verb is fixed, not computed.
+# Custom/plugin/MCP tools have no entry and fall back to the raw preview.
+# =========================================================================
+
+# Each entry maps a built-in tool name to its present-participle verb phrase.
+# A trailing space-then-preview is appended by build_tool_label() when the
+# tool's argument preview is available (e.g. "Reading docs/api.md").
+_TOOL_VERBS: dict[str, str] = {
+ "web_search": "Searching the web",
+ "web_extract": "Reading",
+ "browser_navigate": "Browsing",
+ "browser_click": "Clicking",
+ "browser_type": "Typing",
+ "read_file": "Reading",
+ "write_file": "Writing",
+ "patch": "Editing",
+ "search_files": "Searching files",
+ "terminal": "Running",
+ "execute_code": "Running code",
+ "image_generate": "Generating image",
+ "video_generate": "Generating video",
+ "text_to_speech": "Generating speech",
+ "vision_analyze": "Looking at the image",
+ "session_search": "Searching past sessions",
+ "skill_view": "Reading skill",
+ "skills_list": "Listing skills",
+ "skill_manage": "Updating skill",
+ "delegate_task": "Delegating",
+ "cronjob": "Scheduling",
+ "clarify": "Asking",
+ "memory": "Updating memory",
+ "todo": "Updating tasks",
+}
+
+# Verbs that read better without the raw argument preview appended.
+_TOOL_VERBS_NO_PREVIEW: frozenset[str] = frozenset({
+ "skills_list",
+ "session_search",
+})
+
+# Verbs that take a "for" connector before the preview (search-style phrasing):
+# "Searching the web for " reads better than "Searching the web ".
+_TOOL_VERBS_FOR_CONNECTOR: frozenset[str] = frozenset({
+ "web_search",
+ "search_files",
+})
+
+_friendly_tool_labels: bool = True
+
+
+def set_friendly_tool_labels(enabled: bool) -> None:
+ """Toggle friendly human-phrased tool labels (display.friendly_tool_labels)."""
+ global _friendly_tool_labels
+ _friendly_tool_labels = bool(enabled)
+
+
+def get_friendly_tool_labels() -> bool:
+ """Return whether friendly tool labels are enabled."""
+ return _friendly_tool_labels
+
+
+def get_tool_verb(tool_name: str) -> str | None:
+ """Return the friendly verb for a built-in tool, or None.
+
+ Returns None when friendly labels are disabled or the tool has no curated
+ verb (custom/plugin/MCP tools). Callers that already hold a computed
+ argument preview can compose ``f"{verb} {preview}"`` themselves; use
+ :func:`tool_verb_connector` to pick the right joiner.
+ """
+ if not _friendly_tool_labels:
+ return None
+ return _TOOL_VERBS.get(tool_name)
+
+
+def tool_verb_connector(tool_name: str) -> str:
+ """Return the connector between a verb and its preview (" for " or " ")."""
+ return " for " if tool_name in _TOOL_VERBS_FOR_CONNECTOR else " "
+
+
+def verb_drops_preview(tool_name: str) -> bool:
+ """Whether the verb should render alone, without the argument preview."""
+ return tool_name in _TOOL_VERBS_NO_PREVIEW
+
+
+def build_tool_label(tool_name: str, args: dict, max_len: int | None = None) -> str | None:
+ """Build a human-phrased status label for a tool call.
+
+ For built-in tools with a known verb (``web_search`` -> "Searching the
+ web for ..."), returns the verb optionally followed by the argument
+ preview. For everything else (custom/plugin/MCP tools, or when friendly
+ labels are disabled) returns the raw preview, so callers can use this as a
+ drop-in replacement for :func:`build_tool_preview`.
+ """
+ if not _friendly_tool_labels:
+ return build_tool_preview(tool_name, args, max_len=max_len)
+
+ verb = _TOOL_VERBS.get(tool_name)
+ if not verb:
+ return build_tool_preview(tool_name, args, max_len=max_len)
+
+ if tool_name in _TOOL_VERBS_NO_PREVIEW:
+ return verb
+
+ preview = build_tool_preview(tool_name, args, max_len=max_len)
+ if not preview:
+ return verb
+ if tool_name in _TOOL_VERBS_FOR_CONNECTOR:
+ return f"{verb} for {preview}"
+ return f"{verb} {preview}"
+
+
# =========================================================================
# Inline diff previews for write actions
# =========================================================================
diff --git a/agent/error_classifier.py b/agent/error_classifier.py
index a64683ba41ea..8111880a7ec6 100644
--- a/agent/error_classifier.py
+++ b/agent/error_classifier.py
@@ -31,6 +31,9 @@ class FailoverReason(enum.Enum):
# Billing / quota
billing = "billing" # 402 or confirmed credit exhaustion — rotate immediately
rate_limit = "rate_limit" # 429 or quota-based throttling — backoff then rotate
+ # Upstream model rate-limited (aggregator 429) — fallback to a different
+ # model, NOT credential rotation. The user's key is healthy.
+ upstream_rate_limit = "upstream_rate_limit"
# Server-side
overloaded = "overloaded" # 503/529 — provider overloaded, backoff
@@ -909,6 +912,22 @@ def _classify_by_status(
FailoverReason.overloaded,
retryable=True,
)
+ # Distinguish an OpenRouter-aggregator upstream 429 (an upstream model
+ # like DeepSeek rate-limited OpenRouter's aggregate traffic) from an
+ # account-level 429 (the user's key is actually throttled). OpenRouter
+ # wraps upstream errors with the outer message "Provider returned
+ # error" — the user's key is healthy, so marking it exhausted / rotating
+ # is wrong and burns the key for ~24min. Fall back to a different model.
+ if _is_openrouter_upstream_error(body, provider):
+ upstream_provider = _extract_upstream_provider_name(body)
+ ctx = {"upstream_provider": upstream_provider} if upstream_provider else {}
+ return result_fn(
+ FailoverReason.upstream_rate_limit,
+ retryable=True,
+ should_rotate_credential=False,
+ should_fallback=True,
+ error_context=ctx,
+ )
return result_fn(
FailoverReason.rate_limit,
retryable=True,
@@ -1445,3 +1464,49 @@ def _extract_message(error: Exception, body: dict) -> str:
return msg.strip()[:500]
# Fallback to str(error)
return str(error)[:500]
+
+
+def _is_openrouter_upstream_error(body: Any, provider: str) -> bool:
+ """Detect OpenRouter's aggregator-wrapped upstream provider errors.
+
+ OpenRouter returns errors from upstream model providers (DeepSeek,
+ Anthropic, etc.) wrapped with the outer message "Provider returned error"
+ and the real error nested in ``metadata.raw``. This signal means the
+ user's OpenRouter key is healthy — the upstream provider is the one that
+ failed — so credential rotation is the wrong recovery.
+ """
+ if not isinstance(body, dict):
+ return False
+ provider_lower = (provider or "").strip().lower()
+ err = body.get("error")
+ if not isinstance(err, dict):
+ return False
+ outer_msg = str(err.get("message") or "").strip().lower()
+ if outer_msg != "provider returned error":
+ return False
+ # Require either the explicit OpenRouter provider OR the metadata shape
+ # that only OpenRouter produces (metadata.raw / metadata.provider_name).
+ if provider_lower == "openrouter":
+ return True
+ metadata = err.get("metadata")
+ if isinstance(metadata, dict) and (
+ "raw" in metadata or "provider_name" in metadata
+ ):
+ return True
+ return False
+
+
+def _extract_upstream_provider_name(body: Any) -> Optional[str]:
+ """Pull the upstream provider name out of OpenRouter's error metadata."""
+ if not isinstance(body, dict):
+ return None
+ err = body.get("error")
+ if not isinstance(err, dict):
+ return None
+ metadata = err.get("metadata")
+ if not isinstance(metadata, dict):
+ return None
+ name = metadata.get("provider_name")
+ if isinstance(name, str) and name.strip():
+ return name.strip()
+ return None
diff --git a/agent/gemini_native_adapter.py b/agent/gemini_native_adapter.py
index a79effebba46..c254bf61311b 100644
--- a/agent/gemini_native_adapter.py
+++ b/agent/gemini_native_adapter.py
@@ -337,6 +337,22 @@ def _build_gemini_contents(messages: List[Dict[str, Any]]) -> tuple[List[Dict[st
if parts:
contents.append({"role": gemini_role, "parts": parts})
+ # Gemini's generateContent requires strict user/model alternation;
+ # consecutive same-role contents are rejected with HTTP 400 "Please ensure
+ # that multiturn requests alternate between user and model". The loop above
+ # emits one content per source message, so parallel tool calls (N tool
+ # results become N user functionResponse contents), back-to-back user turns,
+ # or merged assistant turns would each violate that. Merge adjacent
+ # same-role contents by concatenating their parts. For parallel calls this
+ # also produces the grouped multi-functionResponse turn Gemini expects.
+ merged_contents: List[Dict[str, Any]] = []
+ for content in contents:
+ if merged_contents and merged_contents[-1]["role"] == content["role"]:
+ merged_contents[-1]["parts"].extend(content["parts"])
+ else:
+ merged_contents.append(content)
+ contents = merged_contents
+
system_instruction = None
joined_system = "\n".join(part for part in system_text_parts if part).strip()
if joined_system:
diff --git a/agent/learning_graph.py b/agent/learning_graph.py
new file mode 100644
index 000000000000..6dc518b2abaf
--- /dev/null
+++ b/agent/learning_graph.py
@@ -0,0 +1,320 @@
+"""Assemble the "learning made visible" graph for desktop.
+
+This graph is intentionally scoped to what a user actually learns over time:
+- non-base, learned/profile skills (agent-created or used),
+- memory chunks from ``MEMORY.md`` / ``USER.md`` as first-class nodes.
+
+Skill links come from declared ``related_skills``. Memory-to-skill links are
+derived from lexical overlap so the graph can answer "which learned skills are
+connected to the things I remember?".
+
+Run as a module to print edge-density stats against real data:
+
+ python -m agent.learning_graph
+"""
+
+from __future__ import annotations
+
+import json
+import re
+from dataclasses import dataclass, field
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any, Optional
+
+from hermes_constants import get_hermes_home
+
+
+@dataclass
+class SkillNode:
+ name: str
+ category: str
+ source: str = "profile"
+ timestamp: Optional[int] = None
+ use_count: int = 0
+ state: str = "active"
+ created_by: Optional[str] = None
+ pinned: bool = False
+ related: list[str] = field(default_factory=list)
+
+
+def _frontmatter(text: str) -> dict[str, Any]:
+ try:
+ from agent.skill_utils import parse_frontmatter
+
+ fm, _ = parse_frontmatter(text)
+ return fm or {}
+ except Exception:
+ return {}
+
+
+def _related(fm: dict[str, Any]) -> list[str]:
+ raw = fm.get("related_skills") or (fm.get("metadata", {}).get("hermes", {}) or {}).get("related_skills")
+ if isinstance(raw, list):
+ return [str(r).strip() for r in raw if str(r).strip()]
+ if isinstance(raw, str):
+ return [r.strip() for r in raw.strip("[]").split(",") if r.strip()]
+ return []
+
+
+def _category(fm: dict[str, Any], skill_md: Path) -> str:
+ cat = fm.get("category") or (fm.get("metadata", {}).get("hermes", {}) or {}).get("category")
+ if cat:
+ return str(cat)
+ # …/skills///SKILL.md
+ parts = skill_md.parts
+ return parts[-3] if len(parts) >= 3 else "general"
+
+
+def _iter_skill_files(roots: list[tuple[str, Path]]):
+ for source, root in roots:
+ if root.exists():
+ for path in root.rglob("SKILL.md"):
+ yield source, path
+
+
+def _load_usage() -> dict[str, dict[str, Any]]:
+ try:
+ from tools.skill_usage import load_usage
+
+ return load_usage()
+ except Exception:
+ path = get_hermes_home() / "skills" / ".usage.json"
+ try:
+ return json.loads(path.read_text(encoding="utf-8"))
+ except Exception:
+ return {}
+
+
+def _to_int_ts(value: Any) -> Optional[int]:
+ try:
+ if value is None:
+ return None
+ if isinstance(value, (int, float)):
+ return int(value)
+ s = str(value).strip()
+ if not s:
+ return None
+ try:
+ return int(float(s))
+ except ValueError:
+ parsed = datetime.fromisoformat(s.replace("Z", "+00:00"))
+ if parsed.tzinfo is None:
+ parsed = parsed.replace(tzinfo=timezone.utc)
+ return int(parsed.timestamp())
+ except Exception:
+ return None
+
+
+def _usage_timestamp(rec: dict[str, Any]) -> Optional[int]:
+ for key in ("last_activity_at", "last_used_at", "last_viewed_at", "last_patched_at", "created_at"):
+ ts = _to_int_ts(rec.get(key))
+ if ts is not None:
+ return ts
+ return None
+
+
+def build_skill_nodes(skill_roots: list[tuple[str, Path]]) -> dict[str, SkillNode]:
+ usage = _load_usage()
+ nodes: dict[str, SkillNode] = {}
+
+ for source, skill_md in _iter_skill_files(skill_roots):
+ if any(p in {".archive", ".hub", "node_modules", ".git"} for p in skill_md.parts):
+ continue
+ try:
+ fm = _frontmatter(skill_md.read_text(encoding="utf-8")[:4000])
+ except OSError:
+ continue
+ name = str(fm.get("name") or skill_md.parent.name).strip()
+ if not name or name in nodes:
+ continue
+ rec = usage.get(name, {})
+ last_activity = _usage_timestamp(rec)
+ file_ts = _to_int_ts(skill_md.stat().st_mtime)
+ nodes[name] = SkillNode(
+ name=name,
+ category=_category(fm, skill_md),
+ source=source,
+ timestamp=last_activity or file_ts,
+ use_count=int(rec.get("use_count", 0) or 0),
+ state=str(rec.get("state", "active") or "active"),
+ created_by=rec.get("created_by"),
+ pinned=bool(rec.get("pinned", False)),
+ related=_related(fm),
+ )
+ return nodes
+
+
+def build_edges(nodes: dict[str, SkillNode]) -> list[tuple[str, str]]:
+ """Undirected related_skills edges where BOTH endpoints exist (deduped)."""
+ seen: set[tuple[str, str]] = set()
+ edges: list[tuple[str, str]] = []
+ for node in nodes.values():
+ for target in node.related:
+ if target in nodes and target != node.name:
+ a, b = sorted((node.name, target))
+ key = (a, b)
+ if key not in seen:
+ seen.add(key)
+ edges.append(key)
+ return edges
+
+
+def density_stats(nodes: dict[str, SkillNode], edges: list[tuple[str, str]]) -> dict[str, Any]:
+ linked: set[str] = set()
+ for a, b in edges:
+ linked.add(a)
+ linked.add(b)
+ cats: dict[str, int] = {}
+ for n in nodes.values():
+ cats[n.category] = cats.get(n.category, 0) + 1
+ n = len(nodes) or 1
+ return {
+ "nodes": len(nodes),
+ "related_edges": len(edges),
+ "edges_per_node": round(len(edges) / n, 3),
+ "linked_nodes": len(linked),
+ "isolated_pct": round(100 * (n - len(linked)) / n, 1),
+ "categories": len(cats),
+ "agent_created": sum(1 for x in nodes.values() if x.created_by == "agent"),
+ "used": sum(1 for x in nodes.values() if x.use_count > 0),
+ "top_categories": sorted(cats.items(), key=lambda kv: -kv[1])[:8],
+ }
+
+
+def _memory_cards() -> list[dict[str, Any]]:
+ """Freeform memory as readable cards.
+
+ ``MEMORY.md`` / ``USER.md`` are prose split on bare ``§`` separators; each
+ chunk becomes one card. Every chunk is surfaced — the graph shows everything.
+ """
+ base = get_hermes_home() / "memories"
+ cards: list[dict[str, Any]] = []
+ for fname, source in (("MEMORY.md", "memory"), ("USER.md", "profile")):
+ path = base / fname
+ try:
+ text = path.read_text(encoding="utf-8").strip()
+ file_ts = _to_int_ts(path.stat().st_mtime)
+ except OSError:
+ continue
+ for chunk_idx, chunk in enumerate(c.strip() for c in text.split("\n§\n")):
+ if not chunk:
+ continue
+ first = chunk.splitlines()[0].strip().lstrip("# ").strip()
+ cards.append(
+ {
+ "source": source,
+ "timestamp": file_ts + chunk_idx if file_ts is not None else None,
+ "title": (first[:80] + "…") if len(first) > 80 else first,
+ "body": chunk[:1200],
+ }
+ )
+ return cards
+
+
+def _tokenize(text: str) -> set[str]:
+ return {t for t in re.split(r"[^a-z0-9]+", text.lower()) if len(t) >= 3}
+
+
+def _memory_skill_edges(memory_cards: list[dict[str, Any]], skills: list[SkillNode]) -> list[tuple[str, str]]:
+ edges: list[tuple[str, str]] = []
+ skill_meta = [(s, _tokenize(s.name), s.name.lower()) for s in skills]
+ for idx, card in enumerate(memory_cards):
+ mem_id = f"memory:{card['source']}:{idx}"
+ text = f"{card.get('title', '')}\n{card.get('body', '')}".lower()
+ text_tokens = _tokenize(text)
+ scored: list[tuple[int, str]] = []
+ for skill, tokens, skill_name_lower in skill_meta:
+ score = 0
+ if skill_name_lower in text:
+ score += 6
+ score += len(tokens & text_tokens)
+ if score > 0:
+ scored.append((score, skill.name))
+ scored.sort(key=lambda x: (-x[0], x[1]))
+ for _, skill_name in scored[:4]:
+ edges.append((mem_id, skill_name))
+ return edges
+
+
+def _skill_roots() -> list[tuple[str, Path]]:
+ repo = Path(__file__).resolve().parent.parent
+ home_skills = get_hermes_home() / "skills"
+ return [("base", repo / "skills"), ("profile", home_skills)]
+
+
+def build_learning_graph() -> dict[str, Any]:
+ """Full payload for the desktop learning panel.
+
+ Focus on what is profile-learned and actionable:
+ - skills that are NOT base-installed and show real learning signal
+ (agent-created or used),
+ - memory chunks as first-class graph nodes connected to those learned skills.
+ """
+ all_skills = build_skill_nodes(_skill_roots())
+ learned_skills = {
+ name: node
+ for name, node in all_skills.items()
+ if node.source != "base" and (node.created_by == "agent" or node.use_count > 0)
+ }
+ skill_edges = build_edges(learned_skills)
+ memory_cards = _memory_cards()
+ memory_edges = _memory_skill_edges(memory_cards, list(learned_skills.values()))
+
+ edges = skill_edges + memory_edges
+ clusters: dict[str, int] = {}
+ for node in learned_skills.values():
+ clusters[node.category] = clusters.get(node.category, 0) + 1
+ if memory_cards:
+ clusters["memory"] = len(memory_cards)
+
+ graph_nodes = [
+ {
+ "id": n.name,
+ "label": n.name,
+ "kind": "skill",
+ "timestamp": n.timestamp,
+ "category": n.category,
+ "useCount": n.use_count,
+ "state": n.state,
+ "createdBy": n.created_by,
+ "pinned": n.pinned,
+ }
+ for n in learned_skills.values()
+ ]
+ for i, card in enumerate(memory_cards):
+ graph_nodes.append(
+ {
+ "id": f"memory:{card['source']}:{i}",
+ "label": card["title"],
+ "kind": "memory",
+ "memorySource": card["source"],
+ "timestamp": card.get("timestamp"),
+ "category": "memory",
+ "useCount": 0,
+ "state": "active",
+ "createdBy": "memory",
+ "pinned": False,
+ }
+ )
+
+ return {
+ "nodes": graph_nodes,
+ "edges": [{"source": a, "target": b} for a, b in edges],
+ "clusters": [
+ {"category": c, "count": n}
+ for c, n in sorted(clusters.items(), key=lambda kv: -kv[1])
+ ],
+ "memory": memory_cards,
+ "stats": {
+ **density_stats(learned_skills, skill_edges),
+ "memory_nodes": len(memory_cards),
+ "memory_skill_edges": len(memory_edges),
+ "learned_skills": len(learned_skills),
+ },
+ }
+
+
+if __name__ == "__main__":
+ nodes = build_skill_nodes(_skill_roots())
+ print(json.dumps(density_stats(nodes, build_edges(nodes)), indent=2))
diff --git a/agent/lsp/reporter.py b/agent/lsp/reporter.py
index 0eba96ba1ff9..2be1779ccedb 100644
--- a/agent/lsp/reporter.py
+++ b/agent/lsp/reporter.py
@@ -8,6 +8,7 @@
"""
from __future__ import annotations
+import html
from typing import Any, Dict, List
# Severity-1 only by default — warnings/info/hints would flood the
@@ -18,18 +19,65 @@
MAX_PER_FILE = 20
MAX_TOTAL_CHARS = 4000
+# Per-field caps for diagnostic content sourced from the language server.
+# These bound the length of any single attacker-controlled identifier that
+# can ride into the model's tool output via an LSP diagnostic message.
+MAX_MESSAGE_CHARS = 300
+MAX_CODE_CHARS = 80
+MAX_SOURCE_CHARS = 80
+
+
+def _sanitize_field(value: Any, *, limit: int) -> str:
+ """Make a language-server field safe to embed in a tool-result block.
+
+ Diagnostic ``message``, ``code``, and ``source`` originate from a
+ language server that has just parsed user-controlled source code, so
+ they're untrusted from the agent's point of view. A hostile repo can
+ place instruction-shaped text inside identifier names, type aliases,
+ or import paths so the resulting diagnostic echoes that text back
+ into the ```` block the model reads.
+
+ This helper:
+
+ * Collapses CR/LF so a raw newline can't synthesize a new line in the
+ formatted block.
+ * Drops non-printable ASCII control characters that have no business
+ in a single-line summary.
+ * Caps length per-field so a long identifier can't push past the
+ block boundary.
+ * HTML-escapes ``< > &`` so the result can't close ````
+ early or open a new tag.
+
+ Returns ``""`` for ``None`` / empty so the surrounding format string
+ naturally omits the part (mirrors the prior ``if code not in {None,
+ ""}`` check at call sites).
+ """
+ if value is None:
+ return ""
+ raw = str(value)
+ # Collapse newlines so identifier text with raw \n can't fake new lines.
+ raw = raw.replace("\r", " ").replace("\n", " ")
+ # Drop ASCII control chars; keep regular spaces.
+ raw = "".join(ch for ch in raw if ch == " " or ch.isprintable())
+ raw = raw.strip()[:limit]
+ return html.escape(raw, quote=False)
+
def format_diagnostic(d: Dict[str, Any]) -> str:
- """One-line representation of a single diagnostic."""
+ """One-line representation of a single diagnostic.
+
+ ``message``, ``code``, and ``source`` are sanitized before
+ interpolation — see ``_sanitize_field``.
+ """
sev = SEVERITY_NAMES.get(d.get("severity") or 1, "ERROR")
rng = d.get("range") or {}
start = rng.get("start") or {}
line = int(start.get("line", 0)) + 1
col = int(start.get("character", 0)) + 1
- msg = str(d.get("message") or "").rstrip()
- code = d.get("code")
- code_part = f" [{code}]" if code not in {None, ""} else ""
- source = d.get("source")
+ msg = _sanitize_field(d.get("message"), limit=MAX_MESSAGE_CHARS)
+ code = _sanitize_field(d.get("code"), limit=MAX_CODE_CHARS)
+ code_part = f" [{code}]" if code else ""
+ source = _sanitize_field(d.get("source"), limit=MAX_SOURCE_CHARS)
source_part = f" ({source})" if source else ""
return f"{sev} [{line}:{col}] {msg}{code_part}{source_part}"
@@ -57,7 +105,11 @@ def report_for_file(
body = "\n".join(lines)
if extra > 0:
body += f"\n... and {extra} more"
- return f"\n{body}\n"
+ # quote=True escapes both ``"`` and ``&`` so a crafted file name like
+ # ``foo">