feat(agent): add read-only run status introspection (phase 1) (#7297)

* feat(agent): expose read-only run status tool

* refactor(tools): name the subagent dispatch tool set

Hoist the hardcoded subagent dispatch tool-name literal into _SUBAGENT_DISPATCH_TOOL_NAMES so a future dispatch-tool rename has one place to update. No behavior change: the folded-subagent context_fill guard compares the same two names.

* test(tools): account for run status in collision assembly

* docs: keep script guidance within size budget

README.md

@@ -1506,6 +1506,16 @@ actions remain visible beside the compact sidebar on narrow screens.
 
 ### Skills & Tools
 
+Agents can call the read-only `inspect_run_status` tool to inspect visible run
+token usage, message and tool-call counts, an approximate context-window fill,
+and whether a summary or unfinished to-dos exist. It returns counts only, not
+conversation or to-do text. Token counts depend on provider usage metadata;
+compaction can remove earlier messages, and context fill is unavailable when
+the model has no configured `context_window` or its latest response reports no
+input usage. It is also unavailable after a subagent dispatch because subagent
+usage may be attributed to that response. The result identifies whether a
+current-run boundary is known.
+
 Open **Capability Center** from the workspace sidebar to manage **Plugins**
 (MCP servers and Lark/Feishu integration) and **Skills**. Both catalogs support
 search; skill cards show concise descriptions with full descriptions in a detail

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

@@ -68,6 +68,7 @@ Only standalone tool discovery without a model falls back to the base profile.
 1. **Config-defined tools** - Resolved from `config.yaml` via `resolve_variable()`
 2. **MCP tools** - From enabled MCP servers (lazy initialized, cached with resolved-path + content-signature invalidation)
 3. **Built-in tools**:
+   - `inspect_run_status` - Read-only snapshot of visible run message/tool-call counts, reported token usage, the latest reported model-input/context-window ratio, summary presence and unfinished to-do counts. It reads LangGraph `runtime.state` and the worker-owned pre-run message-ID boundary; it never returns message or to-do content. The scope is `current_run_visible` when that boundary is still present (or the run began with no prior messages), otherwise `visible_state`; in the latter case `tool_calls_this_run` is unknown (`null`). Compaction and providers omitting usage can make counts incomplete, so keep these limitations in the tool description and do not claim exact budgeting. Bind the effective model's configured `context_window` during tool assembly; no model-supplied identity or scope arguments are accepted.
    - `present_files` - Make output files visible to user (only `/mnt/user-data/outputs`); virtual paths use `resolve_runtime_user_id(runtime)` so validation resolves the same user-scoped outputs directory established by `ThreadDataMiddleware`
    - `ask_clarification` - Request clarification (intercepted by ClarificationMiddleware, which preserves text fallback and adds `artifact.human_input` for Web UI Human Input Cards). Beyond free text and single choice, the request-side v2 protocol supports `fields` (structured form card collecting several values at once; field types: text/textarea/number/select/multi_select/checkbox/date, validated and normalized server-side in the middleware — invalid entries are dropped, unknown types degrade to `text`; a standalone multi-select question is a one-field form). Replies stay on the v1 response protocol (`text`/`option`): the form card submits a readable text summary
    - `view_image` - Pass the 20 MiB caller limit to sandbox downloads (see [sandbox download contract](../sandbox/AGENTS.md)). Read image bytes for vision-capable models; both sync and async entry points require `sandbox:execute` before any host or sandbox read. With blob storage enabled, validated bytes are persisted as `kind="viewed-image"` and the checkpoint carries a JSON-safe ref beside the compatibility host path; store initialization/write failures return a generic tool error. Blob-backed model reads validate the ref against the separate image metadata before use and fall through to validated compatibility paths on any store failure. Live sandbox bytes remain the fallback for the same sandbox generation, replacement-sandbox recovery uses only SHA-256-verified synchronized host bytes, and async tool invocation drains blocking reads before cancellation may release the sandbox lease

backend/packages/harness/deerflow/tools/builtins/__init__.py

@@ -1,6 +1,7 @@
 from .background_tasks_tool import cancel_background_task, list_background_tasks
 from .batch_task_tool import batch_status, batch_task, cancel_batch, read_batch_result
 from .clarification_tool import ask_clarification_tool
+from .inspect_run_status_tool import build_inspect_run_status_tool
 from .list_uploaded_files_tool import list_uploaded_files
 from .present_file_tool import present_file_tool
 from .review_skill_package_tool import review_skill_package
@@ -15,6 +16,7 @@ __all__ = [
     "present_file_tool",
     "review_skill_package",
     "ask_clarification_tool",
+    "build_inspect_run_status_tool",
     "view_image_tool",
     "task_tool",
     "batch_task",

backend/packages/harness/deerflow/tools/builtins/inspect_run_status_tool.py

@@ -0,0 +1,115 @@
+"""Read-only, bounded counters for the agent's current visible run state."""
+
+from __future__ import annotations
+
+import json
+from collections.abc import Mapping
+from typing import Any
+
+from langchain.tools import BaseTool, tool
+from langchain_core.messages import AIMessage
+
+from deerflow.runtime.context_keys import CURRENT_RUN_PRE_EXISTING_MESSAGE_IDS_KEY
+from deerflow.tools.types import Runtime
+
+_SUBAGENT_DISPATCH_TOOL_NAMES = frozenset({"task", "batch_task"})
+
+
+def _nonnegative_int(value: object) -> int:
+    return value if isinstance(value, int) and not isinstance(value, bool) and value >= 0 else 0
+
+
+def _current_messages(messages: list, context: Mapping[str, Any]) -> tuple[list, str]:
+    """Use the worker's pre-run message IDs to exclude prior turns."""
+    if CURRENT_RUN_PRE_EXISTING_MESSAGE_IDS_KEY not in context:
+        return messages, "visible_state"
+    previous = context[CURRENT_RUN_PRE_EXISTING_MESSAGE_IDS_KEY]
+    if not isinstance(previous, (set, frozenset)) or any(not isinstance(value, str) for value in previous):
+        return messages, "visible_state"
+    if not previous:
+        return messages, "current_run_visible"
+    for index in range(len(messages) - 1, -1, -1):
+        if getattr(messages[index], "id", None) in previous:
+            return messages[index + 1 :], "current_run_visible"
+    # Compaction or checkpoint rewriting can remove every pre-run ID. Do not
+    # label an uncertain count as a run-scoped count.
+    return messages, "visible_state"
+
+
+def snapshot_run_status(state: Mapping[str, Any], context: Mapping[str, Any], *, context_window: int | None) -> dict[str, Any]:
+    """Build a content-free snapshot; no message or todo text leaves the tool."""
+    all_messages = state.get("messages")
+    messages, scope = _current_messages(list(all_messages) if isinstance(all_messages, (list, tuple)) else [], context)
+    input_tokens = output_tokens = total_tokens = missing_usage = tool_calls = 0
+    last_input: int | None = None
+    for message in messages:
+        if not isinstance(message, AIMessage):
+            continue
+        calls = getattr(message, "tool_calls", None)
+        if isinstance(calls, list):
+            tool_calls += len(calls)
+        usage = getattr(message, "usage_metadata", None)
+        if not isinstance(usage, Mapping) or not usage:
+            missing_usage += 1
+            last_input = None
+            continue
+        raw_input = usage.get("input_tokens")
+        input_value = _nonnegative_int(raw_input)
+        output_value = _nonnegative_int(usage.get("output_tokens"))
+        input_tokens += input_value
+        output_tokens += output_value
+        total_tokens += _nonnegative_int(usage.get("total_tokens")) or input_value + output_value
+        last_input = input_value if isinstance(raw_input, int) and not isinstance(raw_input, bool) and raw_input >= 0 else None
+        # TokenUsageMiddleware can fold a completed subagent's usage into the
+        # dispatch AIMessage. Its input count is no longer a context-size proxy.
+        if any(isinstance(call, Mapping) and call.get("name") in _SUBAGENT_DISPATCH_TOOL_NAMES for call in (calls or [])):
+            last_input = None
+
+    fill = None
+    if isinstance(context_window, int) and not isinstance(context_window, bool) and context_window > 0 and last_input is not None:
+        fill = {
+            "used_input_tokens": last_input,
+            "limit_tokens": context_window,
+            "fraction": round(last_input / context_window, 3),
+            "basis": "last_reported_model_input",
+        }
+
+    todos = state.get("todos")
+    pending = in_progress = 0
+    if isinstance(todos, list):
+        for item in todos:
+            if not isinstance(item, Mapping):
+                continue
+            if item.get("status") == "pending":
+                pending += 1
+            elif item.get("status") == "in_progress":
+                in_progress += 1
+    summary = state.get("summary_text")
+    return {
+        "scope": scope,
+        "token_usage": {"input": input_tokens, "output": output_tokens, "total": total_tokens},
+        "usage_missing_messages": missing_usage,
+        "message_count": len(messages),
+        "tool_calls_this_run": tool_calls if scope == "current_run_visible" else None,
+        "context_fill": fill,
+        "has_summary": isinstance(summary, str) and bool(summary.strip()),
+        "todos": {"pending": pending, "in_progress": in_progress},
+    }
+
+
+def build_inspect_run_status_tool(*, context_window: int | None) -> BaseTool:
+    """Bind the effective model's configured context size at assembly time."""
+
+    @tool("inspect_run_status")
+    def inspect_run_status(runtime: Runtime) -> str:
+        """Inspect visible run usage, context fill, and progress counters.
+
+        Read-only; returns no conversation text. Counts may omit compacted
+        messages or provider-omitted usage. Context fill needs a configured
+        window and an unaggregated model input report.
+        """
+        context = runtime.context if isinstance(runtime.context, Mapping) else {}
+        state = runtime.state if isinstance(runtime.state, Mapping) else {}
+        return json.dumps(snapshot_run_status(state, context, context_window=context_window), separators=(",", ":"))
+
+    return inspect_run_status

backend/packages/harness/deerflow/tools/tools.py

@@ -16,6 +16,7 @@ from deerflow.scheduler.runtime import SchedulerRunCapability, is_scheduler_capa
 from deerflow.subagents.batch_runtime import is_subagent_batch_runtime_available
 from deerflow.tools.builtins import (
     ask_clarification_tool,
+    build_inspect_run_status_tool,
     cancel_background_task,
     list_background_tasks,
     list_uploaded_files,
@@ -208,6 +209,7 @@ def get_available_tools(
 
     # Add view_image_tool only if the model supports vision
     model_config = config.get_model_config(model_name) if model_name else None
+    builtin_tools.append(build_inspect_run_status_tool(context_window=model_config.context_window if model_config else None))
     if model_config is not None and model_config.supports_vision:
         builtin_tools.append(view_image_tool)
         logger.info(f"Including view_image_tool for model '{model_name}' (supports_vision=True)")

backend/tests/test_batch_extension_snapshot.py

@@ -65,7 +65,9 @@ async def test_batch_keeps_its_construction_snapshot(monkeypatch, _subagent_env,
     def assemble(**kwargs):
         kwargs["include_mcp"] = False
         tools = real_assembly(**kwargs)
-        captured["tool_names"] = [t.name for t in tools]
+        # This test isolates extension snapshots; the run-status built-in is
+        # assembled independently of the monkeypatched BUILTIN_TOOLS list.
+        captured["tool_names"] = [t.name for t in tools if t.name != "inspect_run_status"]
         return tools
 
     monkeypatch.setattr("deerflow.tools.get_available_tools", assemble)

backend/tests/test_inspect_run_status.py

@@ -0,0 +1,145 @@
+"""Offline contract tests for the read-only run-status tool (issue #7062)."""
+
+import asyncio
+import json
+
+from langchain_core.messages import AIMessage, HumanMessage, ToolMessage
+
+from deerflow.runtime.context_keys import CURRENT_RUN_PRE_EXISTING_MESSAGE_IDS_KEY
+from deerflow.tools.builtins.inspect_run_status_tool import build_inspect_run_status_tool, snapshot_run_status
+
+
+def test_snapshot_counts_only_the_current_run_and_uses_reported_usage():
+    old = HumanMessage(id="old", content="private old request")
+    current = HumanMessage(id="current", content="private current request")
+    call = AIMessage(
+        id="call",
+        content="",
+        tool_calls=[{"name": "bash", "args": {"command": "echo secret"}, "id": "c1"}],
+        usage_metadata={"input_tokens": 60, "output_tokens": 10, "total_tokens": 70},
+    )
+    result = ToolMessage(id="result", content="private output", tool_call_id="c1")
+    final = AIMessage(id="final", content="private answer", usage_metadata={"input_tokens": 80, "output_tokens": 20, "total_tokens": 100})
+    state = {
+        "messages": [old, current, call, result, final],
+        "summary_text": "private summary",
+        "todos": [{"content": "private todo", "status": "pending"}, {"content": "another", "status": "in_progress"}, {"content": "done", "status": "completed"}],
+    }
+    context = {CURRENT_RUN_PRE_EXISTING_MESSAGE_IDS_KEY: frozenset({"old"})}
+
+    snapshot = snapshot_run_status(state, context, context_window=200)
+
+    assert snapshot["scope"] == "current_run_visible"
+    assert snapshot["token_usage"] == {"input": 140, "output": 30, "total": 170}
+    assert snapshot["message_count"] == 4
+    assert snapshot["tool_calls_this_run"] == 1
+    assert snapshot["context_fill"] == {"used_input_tokens": 80, "limit_tokens": 200, "fraction": 0.4, "basis": "last_reported_model_input"}
+    assert snapshot["has_summary"] is True
+    assert snapshot["todos"] == {"pending": 1, "in_progress": 1}
+    serialized = json.dumps(snapshot)
+    assert "private" not in serialized
+    assert "echo secret" not in serialized
+
+
+def test_snapshot_marks_missing_boundary_and_usage_as_unknown():
+    state = {"messages": [AIMessage(id="a", content="hello"), AIMessage(id="b", content="done")], "todos": None}
+
+    snapshot = snapshot_run_status(state, {}, context_window=None)
+
+    assert snapshot["scope"] == "visible_state"
+    assert snapshot["token_usage"] == {"input": 0, "output": 0, "total": 0}
+    assert snapshot["tool_calls_this_run"] is None
+    assert snapshot["usage_missing_messages"] == 2
+    assert snapshot["context_fill"] is None
+    assert snapshot["has_summary"] is False
+    assert snapshot["todos"] == {"pending": 0, "in_progress": 0}
+
+
+def test_compaction_that_removes_run_boundary_reports_visible_state():
+    state = {"messages": [HumanMessage(id="new", content="visible after compaction")]}
+    context = {CURRENT_RUN_PRE_EXISTING_MESSAGE_IDS_KEY: frozenset({"removed-old-message"})}
+
+    snapshot = snapshot_run_status(state, context, context_window=100)
+
+    assert snapshot["scope"] == "visible_state"
+    assert snapshot["message_count"] == 1
+    assert snapshot["tool_calls_this_run"] is None
+
+
+def test_later_unreported_model_response_makes_context_fill_unknown():
+    messages = [
+        AIMessage(id="reported", content="", usage_metadata={"input_tokens": 60, "output_tokens": 10, "total_tokens": 70}),
+        AIMessage(id="unreported", content="done"),
+    ]
+    snapshot = snapshot_run_status({"messages": messages}, {CURRENT_RUN_PRE_EXISTING_MESSAGE_IDS_KEY: frozenset()}, context_window=100)
+    assert snapshot["context_fill"] is None
+    assert snapshot["usage_missing_messages"] == 1
+
+
+def test_subagent_attribution_does_not_masquerade_as_context_fill():
+    dispatch = AIMessage(
+        content="",
+        tool_calls=[{"name": "task", "args": {"prompt": "work"}, "id": "task-1"}],
+        usage_metadata={"input_tokens": 5000, "output_tokens": 100, "total_tokens": 5100},
+    )
+    snapshot = snapshot_run_status({"messages": [dispatch]}, {}, context_window=4096)
+
+    assert snapshot["token_usage"]["total"] == 5100
+    assert snapshot["context_fill"] is None
+
+
+def test_bound_tool_has_no_model_supplied_arguments():
+    tool = build_inspect_run_status_tool(context_window=4096)
+
+    assert tool.name == "inspect_run_status"
+    assert tool.args == {}
+
+
+def test_tool_reads_live_graph_state_without_exposing_message_content():
+    from langchain.agents import create_agent
+    from langchain_core.language_models.fake_chat_models import GenericFakeChatModel
+
+    from deerflow.agents.thread_state import ThreadState
+
+    class FakeModel(GenericFakeChatModel):
+        def bind_tools(self, tools, **kwargs):
+            return self
+
+    model = FakeModel(
+        messages=iter(
+            [
+                AIMessage(content="", tool_calls=[{"name": "inspect_run_status", "args": {}, "id": "status-1"}], usage_metadata={"input_tokens": 25, "output_tokens": 5, "total_tokens": 30}),
+                AIMessage(content="done"),
+            ]
+        )
+    )
+    graph = create_agent(model=model, tools=[build_inspect_run_status_tool(context_window=100)], state_schema=ThreadState)
+    result = asyncio.run(
+        graph.ainvoke(
+            {"messages": [HumanMessage(content="secret request")]},
+            {"configurable": {"thread_id": "status-test"}},
+            context={CURRENT_RUN_PRE_EXISTING_MESSAGE_IDS_KEY: frozenset()},
+        )
+    )
+
+    tool_output = next(message.content for message in result["messages"] if isinstance(message, ToolMessage))
+    snapshot = json.loads(tool_output)
+    assert snapshot["tool_calls_this_run"] == 1
+    assert snapshot["scope"] == "current_run_visible"
+    assert snapshot["token_usage"] == {"input": 25, "output": 5, "total": 30}
+    assert "secret request" not in tool_output
+
+
+def test_available_tools_registers_introspection_with_model_context_window():
+    from deerflow.config.app_config import AppConfig
+    from deerflow.config.model_config import ModelConfig
+    from deerflow.config.sandbox_config import SandboxConfig
+    from deerflow.tools import get_available_tools
+
+    config = AppConfig(
+        models=[ModelConfig(name="status-model", display_name="status-model", description=None, use="langchain_openai:ChatOpenAI", model="status-model", context_window=8192)],
+        sandbox=SandboxConfig(use="deerflow.sandbox.local:LocalSandboxProvider"),
+    )
+    tools = get_available_tools(include_mcp=False, model_name="status-model", app_config=config)
+
+    assert sum(tool.name == "inspect_run_status" for tool in tools) == 1

backend/tests/test_plugin_tools.py

@@ -134,9 +134,12 @@ def test_assembly_keeps_ordinary_and_unaffected_plugin_tools_on_collision(instal
     monkeypatch.setattr("deerflow.tools.builtins.invoke_acp_agent_tool.build_invoke_acp_agent_tool", lambda agents: ordinary)
 
     result = assembly.get_available_tools(app_config=config, extensions=registry.build(), include_mcp=source == "mcp", include_upload_tool=False)
-    assert [tool.name for tool in result] == [name, plugin_tool_name(healthy.namespace, "search")]
-    assert result[0] is ordinary
-    assert result[0].invoke("hello") == "ordinary result"
+    # Model-bound built-ins are assembled independently of BUILTIN_TOOLS.
+    expected_ordinary = [name, "inspect_run_status"] if source in {"config", "builtin"} else ["inspect_run_status", name]
+    assert [tool.name for tool in result] == [*expected_ordinary, plugin_tool_name(healthy.namespace, "search")]
+    retained = next(tool for tool in result if tool.name == name)
+    assert retained is ordinary
+    assert retained.invoke("hello") == "ordinary result"
     assert "Duplicate tool name" in caplog.text
 
     # A host collision must not weaken the strict plugin-vs-plugin check.

scripts/AGENTS.md

@@ -473,7 +473,7 @@ Memory backend async boundary:
   `failure_policy.read: fail_closed`; other recall errors preserve the existing
   log-and-empty-context behavior.
 
-CI runs these regression tests for every pull request via [.github/workflows/backend-unit-tests.yml](../.github/workflows/backend-unit-tests.yml).
+CI: `.github/workflows/backend-unit-tests.yml` runs these tests on every PR.
 
 Agentic browser sessions are process-local. Browser use is refused when the Gateway runs
 more than one worker process, because ordinary uvicorn worker dispatch does not provide