fix(compaction): build the manual-compaction summarization models off the event loop (#7806)

compact_thread_context built the summarization middleware inline on the
Gateway event loop, and with it the run-profile model and any separately
configured summary model. When the provider reads local credential files
(the Codex provider reads the Codex CLI auth.json while it is constructed),
each POST /api/threads/{thread_id}/compact did synchronous file IO on the
shared loop, once for every model it built, so other requests and streams
waited for it. The route passes no app_config, so the call also read the
live config on the loop first, which stats and hashes config.yaml on every
call. #7767 listed this site in the utils guidance as not covered.

Build the middleware through run_assembly, as #7484 and #7767 do for the
one-shot helpers, skill moderation and the goal evaluator, and read the live
config in that pool too when no app_config is passed. The lead agent already
builds the same middleware there through run_agent's agent factory call.
Model resolution, the memory-flush policy, the disabled, failed and
pending-work outcomes and the written checkpoint are unchanged, and
cancellation during construction neither starts the summary call nor writes
a checkpoint.

tests/blocking_io/test_compaction_model_offloop.py builds real
CodexChatModels under the strict Blockbuster gate. On main, compaction
reported itself disabled because every model build hit BlockingError
(os.stat), and the /compact handler answered 500 because the gate rejected
its config read first.

Co-authored-by: Willem Jiang <willem.jiang@gmail.com>

README.md

@@ -2597,9 +2597,10 @@ Backend response cleanup preserves unrelated tag names such as `<think-tank>` an
 The Web UI composer can polish draft input before sending. The rewrite runs as a short Gateway LLM request using the `input_polish` model configuration, keeps slash skill prefixes such as `/data-analysis`, and only replaces the local draft after the user clicks the polish button; it does not create a thread run or persist a message.
 
 Input polishing, follow-up suggestions, background project-file summaries, the
-LLM skill scanner, thread title generation, and the `/goal` evaluator construct
-their model providers off the Gateway event loop, so local credential reads do
-not stall unrelated requests.
+LLM skill scanner, thread title generation, the `/goal` evaluator, and manual
+context compaction (`/compact`) construct their model providers off the
+Gateway event loop, so local credential reads do not stall unrelated
+requests.
 
 When the agent asks for clarification, the Web UI shows the structured response card but keeps the normal composer available. Users can complete the card or send a free-form chat message to bypass it; that message closes the latest pending clarification and becomes the agent's next input. Accompanying answer text stays outside the execution steps panel, and answering the request keeps previously completed text in the conversation while the agent continues.
 

backend/packages/harness/deerflow/runtime/context_compaction.py

@@ -15,6 +15,7 @@ from deerflow.config.agents_config import validate_agent_name
 from deerflow.config.app_config import AppConfig, get_app_config
 from deerflow.runtime.checkpoint_state import CheckpointStateAccessor
 from deerflow.runtime.context_keys import CHECKPOINT_AGENT_NAME_METADATA_KEY, DEFAULT_AGENT_NAME_METADATA_VALUE
+from deerflow.utils.assembly_io import run_assembly
 
 logger = logging.getLogger(__name__)
 _AGENT_CONFIG_NOT_LOADED = object()
@@ -155,7 +156,9 @@ async def compact_thread_context(
     app_config: AppConfig | None = None,
 ) -> ThreadCompactionResult:
     """Summarize old messages in a thread and write a compacted checkpoint."""
-    resolved_app_config = app_config or get_app_config()
+    # Off the event loop (see utils/AGENTS.md): reading the live config stats
+    # and hashes config.yaml. The route passes no app_config.
+    resolved_app_config = app_config or await run_assembly(get_app_config)
     read_config = {"configurable": {"thread_id": thread_id, "checkpoint_ns": ""}}
     snapshot = await accessor.aget(read_config)
     snapshot_config = snapshot.config or {}
@@ -182,7 +185,10 @@ async def compact_thread_context(
         agent_config=agent_config,
     )
     memory_enabled = binding_known and (checkpoint_agent_name is None or (agent_config is not None and getattr(agent_config, "memory_enabled", True) is not False))
-    middleware = _create_compaction_middleware(
+    # Building the middleware constructs its summarization models, and provider
+    # construction can read local credentials; keep it off the loop too.
+    middleware = await run_assembly(
+        _create_compaction_middleware,
         app_config=resolved_app_config,
         keep=keep,
         run_model_name=run_model_name,

backend/packages/harness/deerflow/utils/AGENTS.md

@@ -124,19 +124,21 @@ One-shot input-polish, follow-up-suggestion, and project-file-summary helpers,
 skill moderation (`scan_skill_content`), the goal evaluator, thread title
 generation, and in-graph context compaction build providers through
 `run_assembly` before awaiting inference; the run worker calls its cached
-evaluator factory the same way. Provider construction can read local
-credentials; keep it off the event loop and the default executor while
-preserving request context. Cancellation during construction must not start
-inference. A scan without `app_config` (in-graph `skill_manage`) reads the live
-config in the same pool, and its fail-closed fallback reuses that config rather
-than reading it again on the loop. Other model construction is separate and
-not covered here: manual compaction (`compact_thread_context` builds the
-summarization middleware and its models inline) and, inside graph runs,
-subagent construction.
-Coverage: `tests/blocking_io/test_oneshot_llm_offloop.py`,
+evaluator factory the same way. Manual compaction (`compact_thread_context`)
+builds its summarization middleware and eager models there too. Provider
+construction can read local credentials; keep it off the event loop and the
+default executor while preserving request context. Cancellation during
+construction must not start inference. A scan without `app_config`
+(in-graph `skill_manage`) reads the live config in the same pool, and its
+fail-closed fallback reuses that config rather than reading it again on the
+loop. Manual compaction without `app_config` (the `/compact` route) reads
+the live config in that pool as well. Subagent construction inside graph runs
+is a separate boundary.
+Coverage includes `tests/blocking_io/test_oneshot_llm_offloop.py`,
 `tests/blocking_io/test_system_model_offloop.py`,
-`tests/blocking_io/test_title_model_offloop.py` and
-`tests/blocking_io/test_summarization_model_offloop.py`.
+`tests/blocking_io/test_title_model_offloop.py`,
+`tests/blocking_io/test_summarization_model_offloop.py` and
+`tests/blocking_io/test_compaction_model_offloop.py`.
 
 `start_run` also resolves the factory through `run_assembly` before admission:
 cold imports cannot occupy the loop or its default pool, and import errors

backend/tests/blocking_io/test_compaction_model_offloop.py

@@ -0,0 +1,244 @@
+"""Manual compaction must not build its summarization models on the loop.
+
+``POST /api/threads/{thread_id}/compact`` resolves the live config and builds
+the summarization middleware right before the summary call. That build
+constructs the run-profile model and any separately configured summary model.
+Provider construction can read local credentials (Codex CLI ``auth.json``); the
+real Codex provider is built here so that IO stays inside the strict gate,
+while inference stays offline.
+"""
+
+import asyncio
+import json
+import threading
+from concurrent.futures import ThreadPoolExecutor
+from types import SimpleNamespace
+from unittest.mock import AsyncMock
+
+import pytest
+from langchain_core.messages import AIMessage, HumanMessage
+from langgraph.checkpoint.memory import InMemorySaver
+from langgraph.types import Overwrite
+
+from app.gateway.routers import threads
+from app.gateway.services import build_checkpoint_state_mutation_accessor
+from deerflow.agents.middlewares import summarization_middleware
+from deerflow.config.app_config import AppConfig, get_app_config, reset_app_config
+from deerflow.config.extensions_config import reset_extensions_config
+from deerflow.models.openai_codex_provider import CodexChatModel
+from deerflow.runtime import RunManager
+from deerflow.runtime.context_compaction import ContextCompactionDisabled, compact_thread_context
+from deerflow.runtime.context_keys import CHECKPOINT_AGENT_NAME_METADATA_KEY, DEFAULT_AGENT_NAME_METADATA_VALUE
+from deerflow.runtime.user_context import get_effective_user_id, reset_current_user, set_current_user
+
+pytestmark = pytest.mark.asyncio
+
+THREAD_ID = "compaction-thread"
+_SUMMARY = "The user asked for a trip plan and got one."
+_KEEP = ("messages", 1)
+# Memory stays off, so no pre-compaction flush runs; test_summarization_memory_flush.py covers that hook.
+_CONFIG_YAML = """\
+sandbox:
+  use: deerflow.sandbox.local:LocalSandboxProvider
+models:
+  - name: fixture-codex
+    model: fixture-model
+    use: deerflow.models.openai_codex_provider:CodexChatModel
+summarization:
+  enabled: true
+memory:
+  enabled: false
+"""
+
+
+def _codex_config(**summarization) -> AppConfig:
+    return AppConfig.model_validate(
+        {
+            "sandbox": {"use": "deerflow.sandbox.local:LocalSandboxProvider"},
+            "models": [
+                {"name": "fixture-codex", "model": "fixture-model", "use": "deerflow.models.openai_codex_provider:CodexChatModel"},
+                {"name": "fixture-summary", "model": "fixture-summary-model", "use": "deerflow.models.openai_codex_provider:CodexChatModel"},
+            ],
+            "summarization": {"enabled": True, **summarization},
+            "memory": {"enabled": False},
+        }
+    )
+
+
+@pytest.fixture
+def codex(tmp_path, monkeypatch):
+    """Real Codex construction (credential file IO included); offline inference."""
+    credentials = tmp_path / "auth.json"
+    credentials.write_text(json.dumps({"access_token": "fixture-token", "account_id": "fixture-account"}), encoding="utf-8")
+    monkeypatch.setenv("CODEX_AUTH_PATH", str(credentials))
+    calls = []
+
+    async def invoke(model, prompt, config=None, **kwargs):
+        calls.append((model, threading.get_ident()))
+        return AIMessage(content=_SUMMARY)
+
+    monkeypatch.setattr(CodexChatModel, "ainvoke", invoke)
+    return calls
+
+
+class _SummaryModel:
+    """A built provider; only its summary call is observed."""
+
+    _llm_type = "fixture"
+
+    def __init__(self) -> None:
+        self.ainvoke = AsyncMock(return_value=AIMessage(content=_SUMMARY))
+
+    def with_config(self, **kwargs):
+        return self
+
+
+def _gateway_request():
+    state = SimpleNamespace(checkpointer=InMemorySaver(), checkpoint_channel_mode="full", store=None, run_manager=RunManager())
+    return SimpleNamespace(app=SimpleNamespace(state=state))
+
+
+async def _seeded_thread():
+    """A finished thread bound to the default agent, plus its compaction accessor."""
+    request = _gateway_request()
+    accessor, config = build_checkpoint_state_mutation_accessor(request, thread_id=THREAD_ID, as_node="seed")
+    config["metadata"] = {CHECKPOINT_AGENT_NAME_METADATA_KEY: DEFAULT_AGENT_NAME_METADATA_VALUE}
+    messages = [HumanMessage(id="h1", content="Plan a trip."), AIMessage(id="a1", content="Here is a plan."), HumanMessage(id="h2", content="Book the hotel.")]
+    await accessor.aupdate(config, {"messages": Overwrite(messages)}, as_node="seed")
+    accessor, config = build_checkpoint_state_mutation_accessor(request, thread_id=THREAD_ID, as_node="manual_compaction")
+    return request, accessor, config
+
+
+@pytest.mark.parametrize(
+    ("summarization", "summary_model"),
+    [({}, "fixture-model"), ({"model_name": "fixture-summary"}, "fixture-summary-model")],
+    ids=["run-model", "configured-summary-model"],
+)
+async def test_manual_compaction_builds_codex_models_off_loop(codex, summarization, summary_model):
+    _, accessor, config = await _seeded_thread()
+
+    result = await compact_thread_context(accessor, THREAD_ID, keep=_KEEP, app_config=_codex_config(**summarization))
+
+    # On the loop, the gate rejects every credential read, no model can be
+    # built, and compaction reports itself disabled (the route's 409).
+    assert (result.compacted, result.removed_message_count, result.preserved_message_count) == (True, 2, 1)
+    snapshot = await accessor.aget(config)
+    assert [message.id for message in snapshot.values["messages"]] == ["h2"]
+    assert snapshot.values["summary_text"] == _SUMMARY
+    assert [(model.model, thread) for model, thread in codex] == [(summary_model, threading.get_ident())]
+
+
+@pytest.mark.parametrize("warm_cache", [False, True], ids=["cold-config", "cached-config"])
+async def test_compact_route_reads_live_config_off_loop(tmp_path, monkeypatch, codex, warm_cache):
+    config_path = tmp_path / "config.yaml"
+    extensions_path = tmp_path / "extensions_config.json"
+
+    def seed_files() -> None:
+        config_path.write_text(_CONFIG_YAML, encoding="utf-8")
+        extensions_path.write_text('{"mcpServers": {}, "skills": {}}', encoding="utf-8")
+
+    await asyncio.to_thread(seed_files)
+    monkeypatch.setenv("DEER_FLOW_HOME", str(tmp_path))
+    monkeypatch.setattr("deerflow.config.paths._paths", None)
+    monkeypatch.setenv("DEER_FLOW_CONFIG_PATH", str(config_path))
+    monkeypatch.setenv("DEER_FLOW_EXTENSIONS_CONFIG_PATH", str(extensions_path))
+    reset_app_config()
+    reset_extensions_config()
+    request, _, _ = await _seeded_thread()
+    try:
+        if warm_cache:
+            await asyncio.to_thread(get_app_config)
+        # The route passes no app_config, so compaction resolves the live
+        # config, which stats and hashes config.yaml even on a cache hit.
+        response = await threads.compact_thread.__wrapped__(THREAD_ID, threads.ThreadCompactRequest(keep={"type": "messages", "value": 1}), request)
+    finally:
+        reset_app_config()
+        reset_extensions_config()
+
+    # On the loop, the gate rejects the config read and the route answers 500.
+    assert (response.compacted, response.removed_message_count, response.preserved_message_count) == (True, 2, 1)
+    assert [(model.model, thread) for model, thread in codex] == [("fixture-model", threading.get_ident())]
+
+
+async def test_construction_preserves_context_and_keeps_default_pool_available(monkeypatch):
+    loop = asyncio.get_running_loop()
+    loop.set_default_executor(ThreadPoolExecutor(max_workers=1))
+    started = asyncio.Event()
+    release = threading.Event()
+    observed = []
+    model = _SummaryModel()
+
+    def construct(**kwargs):
+        observed.append((get_effective_user_id(), threading.get_ident()))
+        loop.call_soon_threadsafe(started.set)
+        if not release.wait(timeout=10):
+            raise TimeoutError("test construction was not released")
+        return model
+
+    monkeypatch.setattr(summarization_middleware, "create_chat_model", construct)
+    _, accessor, _ = await _seeded_thread()
+    token = set_current_user(SimpleNamespace(id="request-owner"))
+    task = asyncio.create_task(compact_thread_context(accessor, THREAD_ID, keep=_KEEP, app_config=_codex_config()))
+    try:
+        await asyncio.wait_for(started.wait(), timeout=5)
+        assert not task.done()
+        assert await asyncio.wait_for(asyncio.to_thread(lambda: "unrelated work"), timeout=5) == "unrelated work"
+    finally:
+        release.set()
+        try:
+            await asyncio.wait_for(task, timeout=5)
+        finally:
+            reset_current_user(token)
+
+    assert task.result().compacted is True
+    assert observed == [("request-owner", observed[0][1])]
+    assert observed[0][1] != threading.get_ident()
+    model.ainvoke.assert_awaited_once()
+
+
+async def test_cancelled_construction_does_not_start_inference_or_write(monkeypatch):
+    loop = asyncio.get_running_loop()
+    started = asyncio.Event()
+    finished = asyncio.Event()
+    release = threading.Event()
+    model = _SummaryModel()
+
+    def construct(**kwargs):
+        loop.call_soon_threadsafe(started.set)
+        try:
+            if not release.wait(timeout=10):
+                raise TimeoutError("test construction was not released")
+            return model
+        finally:
+            loop.call_soon_threadsafe(finished.set)
+
+    monkeypatch.setattr(summarization_middleware, "create_chat_model", construct)
+    _, accessor, config = await _seeded_thread()
+    before = await accessor.aget(config)
+    task = asyncio.create_task(compact_thread_context(accessor, THREAD_ID, keep=_KEEP, app_config=_codex_config()))
+    try:
+        await asyncio.wait_for(started.wait(), timeout=5)
+        task.cancel()
+        with pytest.raises(asyncio.CancelledError):
+            await task
+    finally:
+        release.set()
+        await asyncio.wait_for(finished.wait(), timeout=5)
+
+    model.ainvoke.assert_not_awaited()
+    after = await accessor.aget(config)
+    assert after.config["configurable"]["checkpoint_id"] == before.config["configurable"]["checkpoint_id"]
+    assert [message.id for message in after.values["messages"]] == ["h1", "a1", "h2"]
+
+
+async def test_construction_failure_still_reports_compaction_disabled(monkeypatch):
+    def construct(**kwargs):
+        raise ValueError("fixture credential unavailable")
+
+    monkeypatch.setattr(summarization_middleware, "create_chat_model", construct)
+    _, accessor, config = await _seeded_thread()
+
+    with pytest.raises(ContextCompactionDisabled):
+        await compact_thread_context(accessor, THREAD_ID, keep=_KEEP, app_config=_codex_config())
+    snapshot = await accessor.aget(config)
+    assert [message.id for message in snapshot.values["messages"]] == ["h1", "a1", "h2"]