diff --git a/README.md b/README.md index 8a0be33c..be0e778c 100644 --- a/README.md +++ b/README.md @@ -924,12 +924,12 @@ Watch (leapd-hosted) ── observe → SNR filter → score → Finding ── - **Continuous observation** — a board starts as a `Watch` in `leapd`. Scheduler triggers, manual `/board refresh`, and session-analysis batch thresholds all run the same observe→finding cycle; new findings are persisted, severity-gated, and pushed to browsers over WebSocket. - **Refresh model** — the session board re-analyzes as the conversation accumulates turns, when written workspace artifacts change, or on manual `/board refresh`; `/board pause`/`stop` halt re-analysis until resumed. The chosen **template** only changes rendering, never what is analyzed (always the current session). - **Server-Driven UI** — each scenario is authored as a **YAML template** compiled into a validated **ViewSpec** over a fixed component catalog (cards, tables, charts, timelines, gauges, story panels…). Interactive components talk back through a bidirectional action protocol; unknown component types degrade gracefully, and bespoke visuals use a `Custom` escape hatch. The board never renders arbitrary HTML/JS. -- **View client** — the board connects to `leapd` like the TUI does, with no privileged coupling. The web server is optional (`aiohttp`) and degrades with a clear install hint when absent. +- **View client** — the board connects to `leapd` like the TUI does, with no privileged coupling. The web server (`aiohttp`) ships with the base install, so `leap board` works out of the box. ### Install & enable ```bash -pip install 'leapflow[dashboard]' # adds the optional aiohttp web server +pip install leapflow # the aiohttp web server ships with the base install ``` The board binds to `127.0.0.1` with a per-session access token. Tune it through `leap config`: @@ -1381,6 +1381,7 @@ Transport: stdio (JSON-RPC over stdin/stdout) by default. The `PlatformClient` i Apache 2.0 — see [LICENSE](LICENSE). + ---
@@ -1394,4 +1395,4 @@ Apache 2.0 — see [LICENSE](LICENSE).

❤️ Thanks for Visiting ✨ LeapFlow !

Views -

\ No newline at end of file +

diff --git a/pyproject.toml b/pyproject.toml index 7e83b893..ab6d36c5 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -36,7 +36,10 @@ dependencies = [ "rich>=13.0", "prompt_toolkit>=3.0.40", "pyobjc-framework-Quartz>=12.2; sys_platform == 'darwin'", - "pynput>=1.8.0" + "pynput>=1.8.0", + # LeapBoard web server. Part of the core surface (`leap board`), so it ships + # with the base install rather than behind an extra. + "aiohttp>=3.9" ] [project.optional-dependencies] @@ -50,7 +53,6 @@ dev = [ "pytest-cov>=5.0", ] hub = ["modelscope-hub>=0.1.0"] -dashboard = ["aiohttp>=3.9"] # Better main-content extraction for web_fetch. Optional because the stdlib # extractor always ships: this upgrades quality, it does not enable the feature. web = ["trafilatura>=2.2"] diff --git a/src/leapflow/cli/commands/dashboard.py b/src/leapflow/cli/commands/dashboard.py index 8a6f04a4..74804ed1 100644 --- a/src/leapflow/cli/commands/dashboard.py +++ b/src/leapflow/cli/commands/dashboard.py @@ -20,8 +20,8 @@ from leapflow.dashboard import launcher _DEP_HINT = ( - "The dashboard web server requires the optional 'aiohttp' dependency.\n" - "Install it with: pip install 'leapflow[dashboard]'" + "The dashboard web server requires 'aiohttp', which ships with LeapFlow.\n" + "Reinstall it with: pip install aiohttp" ) diff --git a/src/leapflow/dashboard/launcher.py b/src/leapflow/dashboard/launcher.py index a49df706..40c91393 100644 --- a/src/leapflow/dashboard/launcher.py +++ b/src/leapflow/dashboard/launcher.py @@ -32,7 +32,11 @@ def aiohttp_available() -> bool: - """Return True when the optional ``aiohttp`` dependency is importable.""" + """Return True when ``aiohttp`` is importable. + + ``aiohttp`` ships as a core dependency, so this only guards against a + broken or partial install rather than an intentionally omitted extra. + """ import importlib.util return importlib.util.find_spec("aiohttp") is not None @@ -303,8 +307,8 @@ def ensure_server(settings: Any, *, wait_s: float = 8.0) -> dict[str, Any]: return existing if not aiohttp_available(): raise RuntimeError( - "The dashboard web server requires the optional 'aiohttp' dependency. " - "Install it with: pip install 'leapflow[dashboard]'" + "The dashboard web server requires 'aiohttp', which ships with LeapFlow. " + "Reinstall it with: pip install aiohttp" ) # A prior server may be dead-but-recorded, alive with a token we can no longer diff --git a/src/leapflow/dashboard/server.py b/src/leapflow/dashboard/server.py index 4babc53d..3aeb237f 100644 --- a/src/leapflow/dashboard/server.py +++ b/src/leapflow/dashboard/server.py @@ -1,5 +1,5 @@ # Copyright (c) Alibaba, Inc. and its affiliates. -"""Local dashboard web server (optional aiohttp transport, view-client process). +"""Local dashboard web server (aiohttp transport, view-client process). Holds one upstream subscription to the daemon (via DaemonClient) and fans out monitor events to browser WebSockets through a ``ViewHub``. Serves the SDUI @@ -141,7 +141,7 @@ def __init__( # ── App wiring ───────────────────────────────────────────────────────── def build_app(self) -> Any: - """Build the aiohttp Application (requires the optional aiohttp dep).""" + """Build the aiohttp Application.""" from aiohttp import web app = web.Application() diff --git a/src/leapflow/engine/engine.py b/src/leapflow/engine/engine.py index 5fbe8663..6dc9420d 100644 --- a/src/leapflow/engine/engine.py +++ b/src/leapflow/engine/engine.py @@ -399,6 +399,20 @@ def _unknown_tool_retry_prompt(result: Dict[str, Any]) -> str: "right after startup) \u2014 please resend your message." ) +# Injected for the single tool-free round that runs when the loop stops before +# the model has written an answer (a detected repetition loop or an exhausted +# iteration budget). Breaking cold otherwise leaves the user with a generic +# "reasoning step limit" notice and none of the information the tools already +# returned; this asks the model to answer from what it has, with tools withheld +# so it cannot resume the loop. +_FORCED_FINALIZE_PROMPT = ( + "SYSTEM: No further tool calls are available for this turn. Do not attempt " + "to call any tool. Answer the user's request directly and concisely using " + "the information already gathered above. If part of it cannot be determined " + "from what you have, say so plainly and state what would be needed \u2014 do " + "not repeat an earlier tool call." +) + def _is_permission_failure_payload(payload: Dict[str, Any]) -> bool: """Return whether a tool-result payload represents an unresolved permission failure.""" @@ -1630,9 +1644,17 @@ def _check_guardrail( # and the finalize/diversify nudge is suppressed so legitimate batch or # sequential work on a long task is not cut short. Only when the task is # ALSO stalled does the guardrail escalate to a halt (or emit a nudge). + # + # The one exception is a ``progress_independent`` halt: it is raised only + # when the violation is definitionally zero progress (the same tool + # returned the same result N times), so it is honoured regardless of the + # coarse global stall marker -- which a simple factual query may never + # trip, leaving a genuine no-op loop to spin until the budget is spent. frame = self._active_frame stalled = bool(frame is not None and getattr(frame, "stalled_rounds", 0) >= 1) - if violation.severity == "halt" and stalled: + if violation.severity == "halt" and ( + getattr(violation, "progress_independent", False) or stalled + ): messages.append( build_user_message_text( f"SYSTEM GUARDRAIL: {violation.reason}. {violation.suggestion}" @@ -3929,9 +3951,13 @@ async def _run_agent_loop(self, frame: AgentLoopFrame) -> str: if fatal_error: self._emit_chat_event("response", {"content": fatal_error[:500]}) return fatal_error + # The loop stopped without a written answer (repetition halt or exhausted + # budget). Give the model one tool-free round to answer from what it + # gathered before falling back to the canned notice. fallback = ( _app_onboarding_recovery_message(messages) or _last_tool_failures_recovery_message(messages) + or await self._synthesize_forced_answer(messages) or self._budget_exhausted_response(messages) ) self._emit_chat_event("response", {"content": fallback[:500]}) @@ -4282,6 +4308,9 @@ async def _unified_tool_loop_stream( yield StreamEvent(type="thinking", content=content) # Preamble exclusion: content alongside tool_calls is ephemeral # reasoning — exclude from context to prevent final-answer repetition. + # Clear the local copy too: on a later halt/break this must not + # leak as the turn's final answer ahead of a synthesized one. + content = "" assistant_msg: Dict[str, Any] = {"role": "assistant", "content": ""} assistant_msg["tool_calls"] = [ { @@ -4786,10 +4815,15 @@ async def _unified_tool_loop_stream( self._emit_chat_event("response", {"content": final[:500]}) yield StreamEvent(type="final", content=final) else: + # The loop stopped without a written answer (repetition halt or + # exhausted budget). Give the model one tool-free round to answer + # from what it gathered before falling back to the canned notice; + # a genuine terminal failure (fatal_error) still surfaces first. fallback = ( _app_onboarding_recovery_message(messages) or _last_tool_failures_recovery_message(messages) or fatal_error + or await self._synthesize_forced_answer(messages) or self._budget_exhausted_response(messages) ) self._emit_chat_event("response", {"content": fallback[:500]}) @@ -5691,6 +5725,37 @@ def _observe_capability_result(self, result: Any) -> None: # ── Helpers ────────────────────────────────────────────────────────── + async def _synthesize_forced_answer(self, messages: List[Dict[str, Any]]) -> str: + """One tool-free LLM round to answer with what the turn already gathered. + + A turn can stop before the model has written a final answer: a detected + repetition loop is halted, or the iteration budget runs out. Breaking + cold then hands the user a canned "reasoning step limit" notice and none + of the information the tools already returned. This gives the model + exactly one chance to answer from the accumulated context, with tools + withheld so it cannot resume the loop. + + Returns the answer text, or "" on any failure so the caller falls back to + the canned notice \u2014 a best-effort finalize must never turn a clean + stop into a crash. + """ + try: + prompt = list(messages) + prompt.append(build_user_message_text(_FORCED_FINALIZE_PROMPT)) + compressed = self._prepare_llm_messages( + self._healer.heal(prompt), tools=None, round_number=0 + ) + resp = await self._llm.achat( + compressed, stream=False, enable_thinking=False + ) + text = (resp.content or "").strip() + if self._sanitizer: + text = self._sanitizer.sanitize(text) + return text + except Exception: + logger.warning("forced final-answer synthesis failed", exc_info=True) + return "" + @staticmethod def _record_coevolution_resolution(resolution: Any) -> None: """Report one resolution to the co-evolution buffer for the cold-path sweep. diff --git a/src/leapflow/engine/tool_guardrails.py b/src/leapflow/engine/tool_guardrails.py index 041ede2e..5dc00096 100644 --- a/src/leapflow/engine/tool_guardrails.py +++ b/src/leapflow/engine/tool_guardrails.py @@ -31,6 +31,12 @@ class GuardrailViolation: reason: str = "" severity: str = "warning" # "warning" | "halt" suggestion: str = "" + # A halt the engine must honour regardless of the global stall marker. + # Set only when the violation is, by construction, zero forward progress + # (e.g. the same tool returned the same result N times), so that the coarse + # research/governance progress heuristic cannot keep a genuine no-op loop + # spinning until the iteration budget is exhausted. + progress_independent: bool = False @runtime_checkable @@ -42,9 +48,16 @@ def reset(self) -> None: ... class RepetitionGuard: - """Detect exact-duplicate tool calls (same name + same arguments hash). - - Triggers when the same tool call appears N+ times consecutively. + """Detect a no-progress loop: the same tool call returning the same result. + + Triggers when a tool call with identical name + arguments *and* an identical + result appears N+ times consecutively. The result is part of the signature + on purpose: a call that returns a *changing* value each time (legitimate + polling of a sensor, a queue, a build status) is genuine progress and must + not be flagged, whereas a call that keeps returning the *same* value is zero + information gain no matter what the global progress heuristic believes. That + is why the resulting halt is ``progress_independent`` — it is safe to honour + without consulting the stall marker. """ def __init__(self, *, max_repeats: int = 3) -> None: @@ -59,11 +72,24 @@ def check(self, history: List[Dict[str, Any]]) -> GuardrailViolation: if not tool_msgs: return GuardrailViolation(violated=False) + # Correlate each native tool call with the result it produced so the + # signature reflects information gain, not just intent. A call whose + # result is not yet in history (or unmatched) contributes an empty + # result signature, degrading gracefully to call-only comparison. + results_by_id: Dict[str, str] = {} + for m in history: + if m.get("role") == "tool": + content = m.get("content", "") + results_by_id[str(m.get("tool_call_id", ""))] = ( + content if isinstance(content, str) else "" + ) + hashes: List[str] = [] for msg in tool_msgs[-self._max_repeats * 2:]: for tc in (msg.get("tool_calls") or []): fn = tc.get("function", {}) - key = f"{fn.get('name', '')}:{fn.get('arguments', '')}" + result_sig = results_by_id.get(str(tc.get("id", "")), "") + key = f"{fn.get('name', '')}:{fn.get('arguments', '')}:{result_sig}" hashes.append(hashlib.md5(key.encode()).hexdigest()[:12]) if len(hashes) >= self._max_repeats: @@ -71,9 +97,13 @@ def check(self, history: List[Dict[str, Any]]) -> GuardrailViolation: if len(set(tail)) == 1: return GuardrailViolation( violated=True, - reason=f"Identical tool call repeated {self._max_repeats} times", + reason=( + f"Identical tool call returned the same result " + f"{self._max_repeats} times" + ), severity="halt", suggestion="Try a different approach or provide the final answer.", + progress_independent=True, ) return GuardrailViolation(violated=False) diff --git a/tests/test_agent_execution.py b/tests/test_agent_execution.py index 3b32d1ec..c38b04b0 100644 --- a/tests/test_agent_execution.py +++ b/tests/test_agent_execution.py @@ -721,6 +721,102 @@ def reset(self): lt.close() +def test_repetition_guard_is_result_aware() -> None: + """Root-cause guard: a repeated tool call only halts when it also returns the + *same* result (zero information gain). A call whose result changes each time + (legitimate polling) is progress and must not be flagged. The no-progress + halt is ``progress_independent`` so the engine honours it without consulting + the coarse global stall marker.""" + from leapflow.engine.tool_guardrails import RepetitionGuard + + def _call(name: str, args: str, cid: int) -> dict: + return { + "role": "assistant", + "tool_calls": [{"id": cid, "function": {"name": name, "arguments": args}}], + } + + def _result(cid: int, content: str) -> dict: + return {"role": "tool", "tool_call_id": cid, "content": content} + + guard = RepetitionGuard(max_repeats=3) + + # Same call + identical result three times -> stuck loop -> halt. + stuck: list = [] + for i in range(3): + stuck.append(_call("hw_list", "{}", i)) + stuck.append(_result(i, '{"ok": true, "count": 1}')) + v = guard.check(stuck) + assert v.violated and v.severity == "halt" and v.progress_independent + + # Same call but a changing result each time (polling) -> not flagged. + polling: list = [] + for i in range(3): + polling.append(_call("hw_read", '{"d": "s"}', 100 + i)) + polling.append(_result(100 + i, '{"ok": true, "value": %d}' % i)) + assert guard.check(polling).violated is False + + +def test_progress_independent_halt_fires_while_progressing() -> None: + """A ``progress_independent`` halt (the same tool returning the same result) + must stop the loop even when the global stall marker still reads as advancing + -- otherwise a genuine no-op loop spins until the iteration budget is spent + and the user gets a canned step-limit notice instead of an answer.""" + from leapflow.engine.agent_loop import AgentLoopFrame + from leapflow.engine.tool_guardrails import GuardrailViolation + + class _NoProgressHaltGuard: + def check(self, history): + return GuardrailViolation( + violated=True, + reason="loop", + severity="halt", + suggestion="stop", + progress_independent=True, + ) + + def reset(self): + pass + + with tempfile.TemporaryDirectory() as td: + engine, lt, _ = _adaptive_engine(td) + try: + engine._guardrail = _NoProgressHaltGuard() + frame = AgentLoopFrame(user_text="x") + engine._active_frame = frame + frame.stalled_rounds = 0 + msgs = [{"role": "user", "content": "x"}] + # Not stalled, yet the halt fires because it is progress-independent. + assert engine._check_guardrail(msgs) == "halt" + finally: + lt.close() + + +def test_synthesize_forced_answer_returns_model_answer() -> None: + """When the loop stops without a written answer, a single tool-free round lets + the model answer from the gathered context instead of emitting the canned + 'reasoning step limit' notice; a synthesis failure degrades to ''.""" + answer = "Only the host machine is registered; no external USB devices are connected." + with tempfile.TemporaryDirectory() as td: + engine, lt, _ = _adaptive_engine(td) + try: + engine._llm = StubLLM([answer]) + msgs = [ + {"role": "user", "content": "any usb devices connected?"}, + {"role": "tool", "tool_call_id": 0, "content": '{"ok": true, "count": 1}'}, + ] + got = asyncio.run(engine._synthesize_forced_answer(msgs)) + assert got == answer + + class _BoomLLM: + async def achat(self, *a, **k): + raise RuntimeError("provider down") + + engine._llm = _BoomLLM() + assert asyncio.run(engine._synthesize_forced_answer(msgs)) == "" + finally: + lt.close() + + def _with_coordinator(engine): from leapflow.engine.recovery_budget import RecoveryBudget from leapflow.engine.recovery_coordinator import RecoveryCoordinator diff --git a/uv.lock b/uv.lock index 9a084c8d..3460ad10 100644 --- a/uv.lock +++ b/uv.lock @@ -1199,6 +1199,7 @@ wheels = [ name = "leapflow" source = { editable = "." } dependencies = [ + { name = "aiohttp" }, { name = "cryptography" }, { name = "duckdb" }, { name = "gnureadline", marker = "sys_platform == 'darwin'" }, @@ -1217,9 +1218,6 @@ dependencies = [ ] [package.optional-dependencies] -dashboard = [ - { name = "aiohttp" }, -] dev = [ { name = "pytest" }, { name = "pytest-asyncio" }, @@ -1241,7 +1239,7 @@ web = [ [package.metadata] requires-dist = [ - { name = "aiohttp", marker = "extra == 'dashboard'", specifier = ">=3.9" }, + { name = "aiohttp", specifier = ">=3.9" }, { name = "cryptography", specifier = ">=42.0" }, { name = "cua-sandbox", marker = "extra == 'leapspace'" }, { name = "duckdb", specifier = ">=1.0.0" }, @@ -1268,7 +1266,7 @@ requires-dist = [ { name = "trafilatura", marker = "extra == 'web'", specifier = ">=2.2" }, { name = "watchdog", specifier = ">=3.0" }, ] -provides-extras = ["dev", "hub", "dashboard", "web", "leapspace"] +provides-extras = ["dev", "hub", "web", "leapspace"] [[package]] name = "lxml"