Files
hermes-agent/hermes_cli/cli_modal_mixin.py
Siddharth Balyan 70f5dc5f46 feat(connectors): the backend API for the desktop Connectors page; connect an app without a chat session (#115191)
* feat(connectors): the backend serves a connector's tool list, cached for 24 hours

The Connectors page opens one app and shows every tool it has. The backend
had no way to read that list.

- `tools/connectors/portal/`: a client for the portal's tool-list route and a
  JSON cache under the Hermes home, one file per portal origin and connector.
  An entry is fresh for 24 hours. After that the read revalidates with the
  stored ETag: 304 keeps the list, 404 deletes the entry, an upstream failure
  serves the stored list marked stale, and a 401 never serves the cache.
- `connectors.tools {slug, refresh}`: account-level, routed by `profile`, no
  chat session. Errors carry a fixed `reason` from one closed set on the rail.
- Every connector model that is not operation state moves into
  `tui_gateway/contracts/connectors.py`. Handlers that no chat session owns
  live in `tui_gateway/methods_connectors_account.py`.

The wire model is tolerant: an unknown facet reads as unclassified and one odd
tool never blanks a connector.

* feat(connectors): catalog, accounts and member tool rules by RPC

The Connectors page needs the app catalog, the connected account of one app,
a way to disconnect it, and the member's own on/off rules. None had an RPC.

- `connectors.catalog`: name, description, category and logo of each app.
- `connectors.accounts`, `connectors.accounts.remove`: read the accounts at
  the tool gateway and remove one by id.
- `connectors.policy.get`: the rule layers that apply to the member, widest
  first. The body is a union on `mode`, so a reader can name who turned a
  tool off.
- `connectors.policy.set`: one change, a union on `type` (the tools of one
  connector, or one connector on or off), with the revision the user saw. A
  stale revision answers `POLICY_CONFLICT`. The backend composes the upstream
  write in one pure function, so no renderer learns the upstream rules.
- Bundled MCP manifests can name their hosted twin with `connector:`, so the
  page can show one card per app.

* feat(connectors): connect an app without a chat session

Every connector RPC took a `session_id`, and a connect that did not come from
the model's tool call minted a link with no watcher. The Connectors page has
no chat session, and its card must flip to connected by itself.

- `connectors.list`, `connectors.connect`, `connectors.operation.status`,
  `connectors.operation.wake` and `connection.respond` take `owner`, a union
  on `type`: `session` (today's behaviour and authorization) or `account`
  (routed by `profile`, authorized by the live transport like `mcp.*`).
  `session_id` is gone from these params; every desktop caller sends `owner`.
- An account connect runs the same operation lifecycle on a background
  thread, under the profile's scope, so the watcher reads the account and
  settles the operation. A second connect for an app that is already
  connecting returns the open operation and mints nothing.
- `connection.update` carries `owner`. An account operation has no session to
  address, so its updates go out on the session-less broadcast path.

* feat(mcp-catalog): eighteen more bundled entries name their hosted connector

A bundled MCP entry and a hosted connector for the same app are one card
on the Connectors page only when the manifest names its hosted twin.
Linear and Notion had the field. These entries get it too: airtable,
asana, attio, calendly, dropbox, figma, railway, supabase, todoist,
betterstack, canva, cloudflare, datadog, intercom, neon, sentry, stripe
and vercel. Atlassian maps to two hosted connectors and Prisma Postgres
is not clearly the same app, so both stay without one.

* refactor(connectors): the account handlers share one gate, one params model and one write table

The six account-level handlers each repeated the availability gate, the
auth catch and the catch-all reply. One decorator now owns that, and each
handler validates its params with its contract model instead of a ladder
of isinstance checks. The five connection RPCs share one guard for the
unexpected-failure reply.

The four write composers for the member rules were the same function
with a different list key and polarity. They are one table now.

The owner union lives in contracts/common.py, so the params side and the
event side stop declaring it twice and the import cycle is gone.

An account operation start carries one event and a flag, so the wait for
the sign-in link blocks instead of polling every 50 ms. run_operation
loses its two account-only parameters; drive_operation is the second
entry point.

Tests: four deleted (they exercised pydantic or the mock), three merged
into tables, two added (a client that still sends the old top-level
session_id is refused; all six account RPCs run off the server loop).
The shared reply helper and the HTTP and managed-client fakes move to
one place each. Comments are one line or gone.

* fix(connectors): a missing tool-list route reads as "unavailable", not "connector gone"

The tool-list read treated every 404 as the portal's "this connector is
not in the catalog" answer. It deleted the cache entry and answered
CONNECTOR_NOT_FOUND, so a page would offer to remove an app that is
connected and works. A portal that does not serve the route yet answers
a bare 404 for every app.

Only the portal's own {"error": "connector_not_found"} means the
connector is gone. Any other 404 is now a tool-list outage: the cached
list is served as stale, or the RPC answers TOOLS_UNAVAILABLE.

* fix(connectors): a connect from the page returns to the app after sign-in

The sign-in link carries a return target only when the session's surface
is the desktop. A chat session binds that surface. An account-owned call
has no chat session, so nothing bound it: the link was minted without a
return target and the browser ended on the portal's done page instead of
coming back to Hermes.

Every account-owned call now runs with the process's own surface bound,
next to its profile scope. The operation thread copies that context, so
the first link and every reissued link carry the return target and the
operation id.

* test(connectors): defer the new connector RPC coverage

The tests for the new account RPCs, the portal client, the tool-list cache
and the rule composer leave this PR and come back in one later change, after
the API is settled. The same was done for #111008.

Kept: the edits that existing tests need because the five connection RPCs
now take `owner` instead of `session_id`, and the rename of the managed
client seam.

Removed: six new test files, their two fakes and the gateway conftest, and
the new cases in test_mcp_catalog.py, test_connectors_gateway_client.py,
gateway-rpc.test.ts and notifications.test.ts. Reverting this commit restores
all of them.

* fix(cli): the connection panel hands the tool thread back at once

The classic CLI's connection callback waited on a queue for the user's first
decision. The operation's watcher starts only after the callback returns, and
the watcher is what polls a hosted account, runs the 300-second deadline and
sees Ctrl+C.

For a hosted connector the panel opens on the sign-in link, where the only
key that filled the queue was Cancel. The account was never polled: the user
signed in, the panel never changed, and Esc reported the app as skipped.
Ctrl+C set the interrupt flag but left the thread parked on the queue, so the
turn never ended.

The callback now opens the panel and returns, as the gateway's callback does
for the desktop and the Ink TUI. The panel's actions already reach the
operation through apply_answer on the UI thread, so the queue is removed. An
install with a form still waits for Connect, because the backend starts no
work for a pending row. Ctrl+C now settles the operation as `interrupt`, and
open rows become `not_connected`.

Checked on the e2e rig with the fake tool gateway: hosted connect completes on
the third status read; Ctrl+C ends the turn and the polling stops; an MCP
install with a plain and a secret field still saves config and both values.

* fix(connectors): "run it again" lives in the library, so the classic CLI can use it

Making a new sign-in link for a failed or expired hosted connector was
implemented only in the JSON-RPC layer (`_reissue`). The classic CLI does not
go through JSON-RPC: its Connect button on a failed row called apply_answer,
which does nothing for a hosted operation because it has no MCP runner. The
panel showed "Waiting…" until the deadline.

`tools.connectors.run.reissue(operation, names)` now holds the checks and the
per-kind action, and returns a refusal reason or None. The gateway maps each
reason to the same JSON-RPC error as before. The CLI calls it for a hosted
row; a refusal is shown on the row. MCP rows keep their path, because Connect
on a failed MCP row re-sends the form values.

Checked on the e2e rig: a scripted failed sign-in, then Connect: a second mint
with `reinitiate: true`, a new link with a new connection id, then connected.

* feat(connectors): the account list and disconnect go through the portal

`connectors.accounts` and `connectors.accounts.remove` called the tool
gateway. They now call the portal's account-management routes
(`GET /api/v1/connectors/accounts`, `DELETE /api/v1/connectors/accounts/{id}`),
which apply the organisation membership checks and write the disconnect audit
row. There is no fallback to the gateway when the portal is unavailable, and a
removal is never retried.

The read of ONE account stays on the gateway (`GET v1/connectors/accounts/{id}`):
the portal has no such route, and the operation watcher polls it once per second.

`ConnectorClient.list_accounts` and `delete_account` are removed. The removed
account's reply model carries `connector`, which both services send.

* fix(connectors): the account RPCs answer what the portal really sends

Checked against the portal source and against the staging and production
services.

- Errors are read from the upstream error code, not the HTTP status. A rule
  write answered 409 for a stale revision and for a user with no organisation;
  both read as "the policy changed". `org_required` is now `ORG_REQUIRED` and
  403 `no_access` is `ORG_ACCESS_DENIED` on every account RPC; only a rejected
  sign-in is `NEEDS_NOUS_AUTH`. `connectors.list` and `connectors.connect` with
  the account owner map these too.
- `connectors.policy.get` and `connectors.policy.set` carry `effective`: the
  portal's own result for this user, with its stamp and without provider or
  subject ids. Nothing is recomputed locally.
- A rule write needs the revision the user saw: `expected_revision` is required
  and must be a revision string; a bad one is refused before any HTTP call.
- A tool row carries `no_auth`; a list without the upstream flag is an invalid
  answer, not `false`.
- `connectors.accounts.remove` returns the app of the removed account. An
  invalid id is `INVALID_PARAMS`.
- The tool-list cache is per signed-in member (a hash of the token's `sub`),
  so two Nous accounts on one profile do not share entries.
- A malformed slug is a local error, not a 404 from a server nobody called.

Live, staging: no revision and a malformed revision refused locally; a good
revision wrote one disabled Gmail tool and returned it in `effective`; the
same revision again answered `POLICY_CONFLICT`; the list row showed the tool;
the restore brought the member rules back to the start. Live, staging and
production, read-only: all 60 tool lists (5483 tools) parse.

* fix(connectors): the operation RPCs match their contract; a settled card cannot start a new link

Found by two adversarial reviews of the RPC layer and its types.

- `connectors.connect` from a chat session with no open operation is refused
  (`UNKNOWN_OPERATION`). It used to call `manage_connections` through the tool
  registry with no card: it made a link nobody watched, returned a reply
  without the required `settled` field, and named an operation that was never
  registered. There is one way into an operation: the agent's call, or the
  account owner's `connectors.connect`. "Run it again" inside an open
  operation is unchanged.
- `connection.update` for a session is routed by session key AND profile; two
  profiles with the same key no longer cross-deliver a sign-in link. The event
  payload gets the same redaction as the RPC replies.
- `connection.respond` runs on the long-handler pool: an approval can start MCP
  OAuth discovery, which blocked every RPC of the gateway while it ran.
- `connectors.list` rows are a closed snake_case model: `connector`, `enabled`,
  `connected`, `connection_status`, `status_reason`, `gateway_disabled_tools`.
  The last one is display data: the gateway enforces the rules, the backend
  only passes the list on. The phantom `name` and `description` are gone, and
  the desktop uses the generated types instead of hand-written copies.
- `tools_listing` (model-only data) no longer rides on `connectors.operation.status`.
- `unavailable` is removed from the target states and settle reasons: nothing
  produces it. The contract generator now fails when a contract enum and its
  domain enum differ.
- `ConnectorErrorReason` is part of the generated TypeScript and OpenRPC.
- The desktop sends `connection.respond` on the socket that holds the session,
  as wake and reissue already did.
- Contract violations are logged every time, at error level.
- An account connect whose prepare step is slow returns the live operation
  instead of an error while the operation keeps running.
- The MCP-manifest `connector` field leaves this PR (it moves to a later one
  on top of the catalog-reader change). `hermes_cli/mcp_catalog.py` and
  `optional-mcps/` are untouched by this PR again.

anti-slop: no net-new findings (15 touched files).

* fix(connectors): the model gets no sign-in link wherever a card exists; side agents cannot connect

The flag that tells the model "a connection card exists" was the session
platform (`== "desktop"`). The Ink TUI and the classic CLI also draw a card,
so there a connector call on an unconnected app handed the model the raw
`connect_url` and told it to pass the link to the user.

- The agent turn now declares how a link can reach the user
  (`tools/connectors/turn.py`): CARD when the agent was built with a
  connection callback, SIDE for a subagent or a background turn, LINK for a
  headless run (`-q`, cron, ACP, api_server, messaging). It is set once per
  tool batch in the agent loop and read by the connector dispatch path, which
  never sees the agent. The session platform decides return-to-app only.
- CARD: the result carries `connect_card_available` and our hint, never the
  link and never the gateway's own hint.
- SIDE: subagents (`delegate_tool`), gateway background turns and the classic
  CLI `/bg` are built with `side_agent=True`. They hold no `manage_connections`
  tool on any path that derives the tool list, and a connector call on an
  unconnected app gets no link, only "report this to the main agent".
- LINK is unchanged.
- The hosted path with no card builds a detached operation, as the MCP path
  does, so no `connection.update` is emitted for an operation no client asked
  for. Names and docstrings that said "off desktop" now say "no card".
- A settled card is dead on the desktop: `reissueConnectionTarget` and
  `respondToConnectionRequest` share one guard and send nothing for a settled
  or unknown operation.
- The model-facing settled result no longer carries `connection_id`; the model
  repeated it to the user.

Shown on the real clients with a real model (rig, fake tool gateway): Ink TUI
and classic CLI get `connect_card_available` and no link, the model opens the
card, the account connects, the retried call succeeds; `-q` still gets the
link; a subagent and a background turn have no `manage_connections` and get
the no-link hint; on the desktop a card settled with Continue has no enabled
control and sends no RPC.

* feat(tools): every call made through tool_search + tool_call shows a real label on all three clients

A bridged call showed as a generic `tool_call` row in the Ink TUI and as
`⚡ tool_call` in the classic CLI, because the display looked the name up in
the tool registry and bridged names are made at run time. The desktop labelled
only batches that were all hosted connector calls, by parsing names itself.

- `tools/tool_labels.py` is the one place that turns a bridged call into a
  label: kind, app, action, emoji and text. Hosted: `connectors__gmail__GMAIL_SEND_EMAIL`
  → "Gmail · send email". MCP: "Linear · list issues". A local deferred tool
  keeps its own emoji, verb and primary-argument preview. A batch gets exactly
  one label per entry, always; an entry with no name gets a generic label.
- Classic CLI: one row per inner call; the duration on the last row; the
  failure text on the row of the call that failed. With friendly labels off
  it prints what it printed before.
- Gateway: tool start, progress and complete events and stored transcript rows
  carry a typed `labels` field. It does not depend on the classic CLI's
  display setting. Clients no longer parse tool names.
- Ink TUI: rows from the labels; the verbose trail keeps Args and Result.
- Desktop: `ConnectorExecution` renders hosted, MCP and mixed turns from the
  labels, one row per call. The labels reach the row under a key no tool
  argument can use. The connect card it drew under a failed tool result is
  gone: after `CONNECTION_REQUIRED` the one way in is the agent's own
  `manage_connections` call.
- `tool_search` and `tool_describe` rows read "Searching tools · <query>" and
  "Reading tool details · N tools".

Shown on the real desktop (video and screenshots), the Ink TUI and the classic
CLI with the rig: hosted rows, MCP rows, a two-entry batch, a failed entry, a
`CONNECTION_REQUIRED` row with no card under it, labels after a reload, and the
desktop rows with the classic CLI setting off.

* fix(connectors): the model can tell "hosted tools unavailable" from "no such tool"; manage_connections routes MCP names correctly

- A failed hosted search or describe used to return nothing, by design, so the
  model saw only local tools and told the user that a connected app was
  missing. The local results are unchanged; when the hosted leg failed, the
  `tool_search` and `tool_describe` results carry
  `connectors: {status: "unavailable", reason: "unreachable" | "sign_in_expired"}`
  and one hint line. A rejected token is `sign_in_expired`; an entitlement
  refusal or a shut gate adds nothing. `tool_describe` no longer lists those
  names under `not_found` next to "search again".
- NS-932. The description now says which side a name belongs to: a bare name
  is a hosted connector account; `mcp: true` only when the user asks for an MCP
  server, a local server or an install, or when the name exists only in the
  catalog; connect and reconnect are hosted verbs, install, enable and
  authorize are MCP verbs. It names the three clients that draw a card.
- A misrouted target is refused with the call that works. Only when the
  gateway does not know the connector (confirmed on that failure path) and the
  name is a catalog entry does the target fail with "X is a local MCP server.
  Call manage_connections with action install ...". It is a per-target
  outcome: other targets of the same call keep their links and their card. A
  vendor failure on a name both sides know stays an ordinary failed row. The
  MCP side mirrors it, and never for an entry that is only not installed.
- "Do not re-ask after a skip or a timeout" no longer stops the model when the
  USER asks for that app again; the description and the settled-result notes
  say so. A builder saw the model refuse a direct user request.

Shown on the Ink TUI and the classic CLI with a real model: a dead gateway and
a 401; "connect fxmail" goes hosted; "install the fx-noauth MCP server" goes
MCP; "connect fx-noauth" reaches the MCP install card in one corrective round
with no hosted mint; a two-target call where one is misrouted still connects
the other with exactly one mint.

* fix(tui): the connection card answers every key, shows what is happening, and is dead once settled

Reproduced on the real Ink TUI with the rig, then fixed:

- The keyboard was dead during the sign-in wait: the card kept a `submitting`
  flag that the normal OAuth path never cleared, and Esc went through the same
  guard. The in-flight state now belongs to the answered row and clears when
  that row moves, when any later frame of the operation arrives, or after
  five seconds. Esc skips the row in every phase; Ctrl+C interrupts the turn
  (the input handler had no branch for this overlay); Shift+arrows scroll the
  transcript and the card ignores them; arrow keys no longer move the text
  cursor and the field focus at once.
- The card was lost at turn idle: the overlay flag was cleared while the
  operation stayed in the store, and a resume dropped the pending card. The
  flag survives idle, a resume shows the pending card again, a session switch
  clears it.
- States with no branch: `not_connected` and a row with no link fell into the
  credential form; `expired` vanished with no note. The title and the row text
  now name the action (connect, reconnect, install, enable, authorize); a
  failed or expired row with no fields offers Try again / Skip; a failed row
  WITH fields reopens the form over the typed draft, with the failure above it.
- A settled card is dead: at settle the overlay closes and one transcript line
  per app states the outcome. A settled or dismissed operation id is
  remembered, so no replay or resume can reopen its card. Esc in the last
  "Finishing…" moment hides the card and still writes the outcome lines.
- A failed `connection.respond` and a browser that did not open are shown on
  the card in one sentence.

Also: `tui_gateway/connector_payload.py` redacted the BOOLEAN `secret` flag of
a credential field to the string "[REDACTED]". On the desktop every credential
field therefore rendered as a password and lost its prefilled default. A
boolean is no longer redacted.

* chore(connectors): remove the comments and docstrings this branch added

Deletions only. Kept: tool directives (`# noqa`, `// eslint-disable`, ...),
`// SAFETY:` lines, and the docstrings of the contract models under
`tui_gateway/contracts/`, which become the descriptions in the generated
OpenRPC and TypeScript.

Checked that no code changed: every Python file has the same AST as before
once docstrings and `pass` are ignored (62 files), and every TypeScript file
prints the same with comments stripped by the TypeScript printer (32 files).
The generated contract files are unchanged.

* fix(connectors): a card restored after a reload answers again; every account RPC names auth and org failures

Found by the end-to-end runs on the pushed head.

- Desktop: after a window reload, Continue on the restored card sent nothing.
  The answer looked up the backend that holds the session with the runtime
  session id, the lookup wants the stored id, and a failed lookup returned
  silently. When the lookup gives no owner the answer now goes out on the
  window's active socket, which is what main does.
- `connectors.policy.get` answered `POLICY_UNAVAILABLE` for a rejected sign-in,
  a refused scope, a non-member and a missing organisation alike: the handler
  runs with the gateway's globals and did not import the reason enum, so its
  own error mapping raised. `connectors.accounts.remove` caught auth failures
  in its generic branch. `org_required` was mapped on `policy.set` only. All
  six account RPCs now answer `NEEDS_NOUS_AUTH`, `FORBIDDEN_SCOPE`,
  `ORG_ACCESS_DENIED` and `ORG_REQUIRED` for those four upstream answers.
2026-09-22 13:57:51 +05:30

1328 lines
59 KiB
Python

"""Modal overlays for the interactive CLI: clarify, approval, sudo/secret capture, command palette,
slash-confirm, external editor.
Mixin split out of ``cli.py``; bound onto ``HermesCLI`` via the MRO. cli.py-internal symbols are
imported LAZILY inside each method — the mixin never imports ``cli`` at module load time (cycle).
"""
from __future__ import annotations
import json
import queue
import sys
import threading
import time as _time
import webbrowser
from hermes_cli.callbacks import prompt_for_secret
from typing import Optional
_TIMED_OUT = object() # sentinel returned by _poll_modal_queue when the deadline passes
# Typed answers accepted by the slash-confirm modal, mapped onto the canonical choice values.
_CONFIRM_ALIASES = {
"1": "once", "once": "once", "approve": "once", "yes": "once", "y": "once", "ok": "once",
"2": "always", "always": "always", "remember": "always",
"3": "cancel", "cancel": "cancel", "nevermind": "cancel", "no": "cancel", "n": "cancel"}
_APPROVAL_OUTCOME_LABELS = {
"once": "allowed once",
"session": "allowed for session",
"always": "added to allowlist",
"deny": "denied"}
_CLARIFY_TIMEOUT_REPLY = (
"The user did not provide a response within the time limit. "
"Use your best judgement to make the choice and proceed.")
def _approval_gate_on(key: str) -> bool:
"""Read ``approvals.<key>`` (default on); any load failure keeps the prompt enabled."""
from cli import load_cli_config
try:
cfg = load_cli_config()
approvals = cfg.get("approvals") if isinstance(cfg, dict) else None
if isinstance(approvals, dict):
return bool(approvals.get(key, True))
except Exception:
pass
return True
def _gated_confirm(self, command, key, *, title, detail, choices, unchanged, always_msg, once_verb):
"""Shared once/always/cancel confirm behind ``approvals.<key>`` (destructive slash, /reload-mcp).
Returns ``"once"`` without prompting when the gate is off; ``None`` on cancel / no input /
unrecognized answer (already reported to the user). Picking "always" persists the opt-out.
"""
from cli import save_config_value
if not _approval_gate_on(key):
return "once"
raw = self._prompt_text_input_modal(title=title, detail=detail, choices=choices)
if raw is None:
print(f"🟡 /{command} cancelled (no input).")
return None
choice = self._normalize_slash_confirm_choice(raw, choices)
if choice is None:
print(f"🟡 Unrecognized choice '{raw}'. /{command} cancelled.")
return None
if choice == "cancel":
print(f"🟡 /{command} cancelled. {unchanged}")
return None
if choice == "always":
if save_config_value(f"approvals.{key}", False):
print(always_msg)
print(f" Re-enable via `approvals.{key}: true` in config.yaml.")
else:
print(f"⚠️ Couldn't persist opt-out — {once_verb} once.")
return choice
class CLIModalMixin:
"""Modal overlays for the interactive CLI: clarify, approval, sudo/secret capture, command
palette, slash-confirm, external editor."""
_connection_state = None
def _open_external_editor(self, buffer=None) -> bool:
"""Open the active input buffer in an external editor."""
from cli import _DIM, _RST, _cprint
app = getattr(self, "_app", None)
if not app:
_cprint(f"{_DIM}External editor is only available inside the interactive CLI.{_RST}")
return False
if self._command_running:
_cprint(f"{_DIM}Wait for the current command to finish before opening the editor.{_RST}")
return False
if (self._sudo_state or self._secret_state or self._approval_state
or getattr(self, "_slash_confirm_state", None) or self._clarify_state
or self._connection_state):
_cprint(f"{_DIM}Finish the active prompt before opening the editor.{_RST}")
return False
target_buffer = buffer or getattr(app, "current_buffer", None)
if target_buffer is None:
_cprint(f"{_DIM}No active input buffer is available for the external editor.{_RST}")
return False
try:
# Inline pastes so the editor sees real content; set the skip flag unconditionally so
# the editor-close text-change doesn't re-collapse it.
self._inline_pastes(target_buffer)
self._skip_paste_collapse = True
# Submission here is driven by the custom `enter` keybinding, NOT the buffer's
# accept_handler, so validate_and_handle can't route through it; chain a done-callback
# that re-uses the real submit pipeline (TUI Ctrl+G parity: save == send).
task = target_buffer.open_in_editor(validate_and_handle=False)
if task is not None and hasattr(task, "add_done_callback"):
task.add_done_callback(lambda _t, b=target_buffer: self._submit_editor_buffer(b))
return True
except Exception as exc:
_cprint(f"{_DIM}Failed to open external editor: {exc}{_RST}")
return False
def _submit_editor_buffer(self, buffer) -> None:
"""Submit the draft an external editor left in ``buffer`` (Ctrl+G done-callback), mirroring
the `enter` keybinding: empty save ignored, bang/slash dispatched, else queued. Runs on the
prompt_toolkit loop, so it must stay cheap/non-blocking."""
from cli import _DIM, _RST, _cprint, _looks_like_slash_command
try:
text = (getattr(buffer, "text", "") or "").strip()
except Exception:
return
if not text:
return
app = getattr(self, "_app", None)
def _done() -> None:
self._reset_input_buffer(buffer)
if app is not None:
app.invalidate()
# `!<command>` shell mode is checked before slash dispatch, matching the Enter path.
try:
if self.handle_bang_shell(text):
_done()
return
except Exception as exc:
_cprint(f" {_DIM}Shell command failed: {exc}{_RST}")
_done()
return
if _looks_like_slash_command(text):
try:
if not self.process_command(text):
self._should_exit = True
if app is not None and app.is_running:
app.exit()
except Exception as exc:
_cprint(f" {_DIM}Command failed: {exc}{_RST}")
finally:
_done()
return
if self._agent_running:
# Agent busy → honour the configured busy-input behaviour (interrupt/steer remain
# reachable via the normal Enter path).
if self.busy_input_mode == "interrupt":
self._interrupt_queue.put(text)
else:
self._pending_input.put(text)
preview = text[:80] + ("..." if len(text) > 80 else "")
_cprint(f" Queued for the next turn: {preview}")
else:
self._pending_input.put(text)
_done()
def _inline_pastes(self, buffer) -> None:
"""Replace collapsed ``[Pasted text #N -> file]`` placeholders in ``buffer`` with real text.
History recall and the external editor need the content (the file may be gone or on another
machine); inlining before ``reset(append_to_history=True)`` also lets prompt_toolkit persist
it. Sets ``_skip_paste_collapse`` so the ensuing text-change doesn't re-collapse it.
"""
from cli import logger
try:
existing = getattr(buffer, "text", "")
expanded = self._expand_paste_references(existing)
if expanded != existing and hasattr(buffer, "text"):
self._skip_paste_collapse = True
buffer.text = expanded
if hasattr(buffer, "cursor_position"):
buffer.cursor_position = len(expanded)
except Exception:
logger.debug("Failed to inline paste placeholders", exc_info=True)
def _reset_input_buffer(self, buffer) -> None:
"""Clear an input buffer after a programmatic submit (best-effort)."""
try:
buffer.reset(append_to_history=True)
except Exception:
try:
buffer.text = ""
except Exception:
pass
def _prefill_input_buffer(self, text: str) -> None:
"""Place ``text`` in the active prompt_toolkit buffer, editable."""
from cli import logger
app = getattr(self, "_app", None)
if app is None:
return
try:
buf = app.current_buffer
buf.text = text
if hasattr(buf, "cursor_position"):
buf.cursor_position = len(text)
app.invalidate()
except Exception as e:
logger.debug("undo: prefill buffer failed: %s", e)
def _prompt_text_input(self, prompt_text: str) -> str | None:
"""Prompt for free-text input safely inside or outside prompt_toolkit.
``run_in_terminal`` only works on the main-thread loop; on the ``process_loop`` daemon
thread a bare ``input()`` would block forever on loop-owned stdin, so with an app running
off-main we cancel cleanly (None) — mirroring ``_stdin_fallback`` in the modal prompt.
Mirrors the thread-aware guard in ``_run_curses_picker``: ``run_in_terminal`` returns a coroutine
that must be awaited by the prompt_toolkit event loop, which only exists on the main thread. Slash
commands are dispatched from the ``process_loop`` daemon thread (see issue #23185), so calling
``run_in_terminal`` from there orphans the coroutine — ``_ask`` never runs, and user keystrokes leak
into the composer instead. Fall back to a direct ``input()`` when we're off the main thread.
"""
result = [None]
def _ask():
try:
result[0] = input(prompt_text).strip() or None
except (KeyboardInterrupt, EOFError):
pass
in_main_thread = threading.current_thread() is threading.main_thread()
# Slash-worker guard (#23185 / billing auto-reload hang): when a prompt_toolkit app is running but
# we're on a non-main thread (the process_loop / TUI slash-worker daemon thread), stdin is owned by
# the event loop / JSON-RPC pipe. A bare input() there blocks forever until the worker's 45s timeout
# fires. We cannot safely prompt off the main thread, so cancel cleanly (None) instead of hanging —
# mirrors the _stdin_fallback discipline in _prompt_text_input_modal.
if self._app and not in_main_thread:
self._invalidate()
return None
if self._app and in_main_thread:
from prompt_toolkit.application import run_in_terminal
was_visible = self._status_bar_visible
self._status_bar_visible = False
self._app.invalidate()
try:
run_in_terminal(_ask)
except Exception:
# WSL / Warp / some emulators silently drop the scheduled coroutine — fall back to
# a direct input() so keystrokes don't leak into the agent buffer.
try:
_ask()
except Exception:
pass
finally:
self._status_bar_visible = was_visible
self._app.invalidate()
else:
_ask()
return result[0]
def _poll_modal_queue(self, response_queue, deadline_attr, *, refresh=1.0, paint=None):
"""Block until a value lands on ``response_queue`` or ``self.<deadline_attr>`` passes
(``None`` deadline = unlimited). Returns the value or ``_TIMED_OUT``.
Repaints every ``refresh`` seconds (``0`` = on every idle tick) so countdown hints stay
live; ``paint`` defaults to ``_paint_now`` — modal prompts must bypass the ``_invalidate``
throttle/resize guard or the panel can be dropped and time out unseen.
"""
paint = paint or self._paint_now
last = _time.monotonic()
while True:
try:
return response_queue.get(timeout=1)
except queue.Empty:
deadline = getattr(self, deadline_attr)
if deadline is not None and deadline - _time.monotonic() <= 0:
return _TIMED_OUT
now = _time.monotonic()
if now - last >= refresh:
last = now
paint()
def _prompt_text_input_modal(
self, *, title: str, detail: str, choices: list[tuple[str, str, str]], timeout: float = 120
) -> str | None:
"""Slash-command confirmation through the prompt_toolkit composer (raw input() fought
prompt_toolkit's stdin ownership: prompt above the TUI, Enter read as EOF). All platforms
drive the modal via ``self._app.loop`` + ``call_soon_threadsafe``; raw ``input()`` is kept
only for the safe cases (no app, no loop, scheduling failure) — on Windows a non-main-thread
input() deadlocks against prompt_toolkit, so that case cancels instead.
**Platform note (Windows — issue #33961):** Earlier code bypassed the modal on ``sys.platform ==
"win32"`` and fell back to a raw ``input()`` prompt. When the confirm was triggered from the
``process_loop`` daemon thread (the normal case) that ``input()`` ran off the main thread and
deadlocked against prompt_toolkit's stdin ownership — the user saw a frozen cursor and Ctrl-C was
swallowed (bare ``/reset`` froze; ``/reset now`` worked only because it skips the prompt entirely).
"""
if not choices:
return None
if not getattr(self, "_app", None):
return self._prompt_text_input("Choice [1/2/3]: ")
try:
app_loop = self._app.loop
except Exception:
app_loop = None
in_main_thread = threading.current_thread() is threading.main_thread()
def _stdin_fallback() -> str | None:
# On native Windows a raw input() from a non-main thread deadlocks against prompt_toolkit's
# stdin ownership (#33961). With an app running we cannot safely prompt off the main thread, so
# cancel cleanly (None) rather than hang the terminal.
if sys.platform == "win32" and not in_main_thread:
self._invalidate()
return None
return self._prompt_text_input("Choice [1/2/3]: ")
if not in_main_thread and app_loop is None:
return _stdin_fallback()
response_queue = queue.Queue()
def _setup_modal() -> None:
self._capture_modal_input_snapshot()
self._slash_confirm_state = {
"title": title,
"detail": detail,
"choices": choices,
"selected": 0,
"response_queue": response_queue}
self._slash_confirm_deadline = _time.monotonic() + timeout
self._invalidate()
def _teardown_modal() -> None:
self._slash_confirm_state = None
self._slash_confirm_deadline = 0
self._restore_modal_input_snapshot()
self._invalidate()
def _run_on_app_loop(fn) -> bool:
if in_main_thread or app_loop is None:
fn()
return True
ready = threading.Event()
def _wrapped() -> None:
try:
fn()
finally:
ready.set()
try:
app_loop.call_soon_threadsafe(_wrapped)
except Exception:
return False
return ready.wait(timeout=5)
if not _run_on_app_loop(_setup_modal):
return _stdin_fallback()
try:
result = self._poll_modal_queue(
response_queue, "_slash_confirm_deadline", refresh=5.0, paint=self._invalidate)
if result is not _TIMED_OUT:
_run_on_app_loop(_teardown_modal)
return result
finally:
if self._slash_confirm_state is not None:
_run_on_app_loop(_teardown_modal)
return None
def _submit_slash_confirm_response(self, value: str | None) -> None:
state = self._slash_confirm_state
if not state:
return
state["response_queue"].put(value)
self._slash_confirm_state = None
self._slash_confirm_deadline = 0
self._invalidate()
def _normalize_slash_confirm_choice(
self, raw: str | None, choices: list[tuple[str, str, str]]) -> str | None:
if raw is None:
return None
choice_raw = raw.strip().lower()
if not choice_raw:
return None
allowed = {choice[0] for choice in choices}
normalized = _CONFIRM_ALIASES.get(choice_raw)
if normalized in allowed:
return normalized
if choice_raw in allowed:
return choice_raw
return None
def _build_command_palette_entries(self) -> list:
"""Flat (command, category, desc) rows for the Ctrl+P palette: the COMMAND_REGISTRY behind
/help filtered to this surface, plus installed skill commands. Selecting inserts the exact
command string — never a fuzzy resolution."""
from cli import _ensure_skill_commands
from hermes_cli.commands import COMMANDS_BY_CATEGORY
entries: list[tuple[str, str, str]] = []
for category, commands in COMMANDS_BY_CATEGORY.items():
for cmd, desc in commands.items():
if self._command_available(cmd):
entries.append((cmd, category, desc))
try:
for cmd, info in sorted(_ensure_skill_commands().items()):
entries.append((cmd, "Skill", info.get("description", "")))
except Exception:
pass
return entries
def _open_command_palette(self) -> None:
"""Open the Ctrl+P fuzzy command palette modal (never stacked over another modal)."""
if getattr(self, "_command_palette_state", None):
return
if (self._model_picker_state or self._clarify_state or self._approval_state
or self._slash_confirm_state or self._sudo_state or self._secret_state
or self._connection_state):
return
self._capture_modal_input_snapshot()
self._command_palette_state = {
"entries": self._build_command_palette_entries(),
"filter": "",
"selected": 0,
"_scroll_offset": 0}
self._invalidate(min_interval=0.0)
def _close_command_palette(self) -> None:
self._command_palette_state = None
self._restore_modal_input_snapshot()
self._invalidate(min_interval=0.0)
def _command_palette_visible_entries(self) -> list:
"""Rows matching the active filter, ranked command-name-first (a bare subsequence over
"cmd category desc" is uselessly permissive — "steer" would match 130+ rows via text):
0 exact command, 1 command startswith, 2 substring in command, 3 subsequence in command,
4 substring in description. Non-matches are dropped; ties keep registry order."""
state = self._command_palette_state or {}
entries = state.get("entries") or []
q = (state.get("filter", "") or "").strip().lower()
if not q:
return list(entries)
def _subseq(needle: str, hay: str) -> bool:
it = iter(hay)
return all(ch in it for ch in needle)
qn = q.lstrip("/")
ranked = []
for order, row in enumerate(entries):
cmd, _cat, desc = row
name = cmd.lower().lstrip("/")
desc_l = (desc or "").lower()
if name == qn:
rank = 0
elif name.startswith(qn):
rank = 1
elif qn in name:
rank = 2
elif _subseq(qn, name):
rank = 3
elif q in desc_l:
rank = 4
else:
continue
ranked.append((rank, order, row))
ranked.sort(key=lambda t: (t[0], t[1]))
return [row for (_r, _o, row) in ranked]
def _handle_command_palette_selection(self) -> None:
"""Prefill the selected command into the composer — never auto-run (many take args)."""
from cli import logger
state = self._command_palette_state
if not state:
return
rows = self._command_palette_visible_entries()
selected = state.get("selected", 0)
if not (0 <= selected < len(rows)):
self._close_command_palette()
return
cmd = rows[selected][0]
self._close_command_palette()
try:
app = getattr(self, "_app", None)
if app is not None:
buf = app.current_buffer
buf.text = cmd + " "
buf.cursor_position = len(buf.text)
self._invalidate(min_interval=0.0)
except Exception:
logger.debug("command palette prefill failed", exc_info=True)
@classmethod
def _split_destructive_skip(cls, cmd_text: Optional[str]) -> tuple[str, bool]:
"""Split inline-skip tokens out of a destructive slash command → ``(remainder, skip)``.
``remainder`` is the text minus the leading "/cmd" word and any skip tokens; ``skip`` is
True iff one was found: "/reset now" -> ("", True); "/reset --yes My title" ->
("My title", True); "/new My title" -> ("My title", False).
"""
tokens = (cmd_text or "").strip().split()
if not tokens:
return "", False
if tokens[0].startswith("/"):
tokens = tokens[1:]
kept = [tok for tok in tokens if tok.lower() not in cls._DESTRUCTIVE_SKIP_TOKENS]
return " ".join(kept), len(kept) != len(tokens)
def _confirm_destructive_slash(
self, command: str, detail: str, cmd_original: Optional[str] = None) -> Optional[str]:
"""Confirm a destructive slash command (``/clear``, ``/new``/``/reset``, ``/undo``): returns
``"once"``, ``"always"`` (persists the opt-out) or ``None`` (cancelled). Gate off → "once"
silently; ``now`` / ``--yes`` / ``-y`` in ``cmd_original`` bypasses the modal (callers strip
the tokens via :meth:`_split_destructive_skip`).
Inline-skip: if ``cmd_original`` contains ``now``, ``--yes``, or ``-y`` as an argument (e.g.
``/reset now``, ``/new --yes My title``), the modal is bypassed and ``"once"`` is returned
immediately. This is an escape hatch for non-interactive use and for the degraded path where the
modal can't be marshaled onto the app loop (native Windows itself now drives the modal normally —
see #33961). Callers are responsible for stripping the skip tokens from any remaining argument
parsing (see :meth:`_split_destructive_skip`).
"""
if cmd_original and self._split_destructive_skip(cmd_original)[1]:
return "once"
return _gated_confirm(
self, command, "destructive_slash_confirm",
title=f"⚠️ /{command} — destroys conversation state",
detail=detail,
choices=[
("once", "Approve Once", "proceed this time only"),
("always", "Always Approve", "proceed and silence this prompt permanently"),
("cancel", "Cancel", "keep current conversation")],
unchanged="Conversation unchanged.",
always_msg="🔒 Future /clear, /new, /reset, and /undo will run without confirmation.",
once_verb="proceeding")
def _ring_bell(self, prompt: bool = False, context: str = "", detail: str = "") -> None:
"""Terminal bell (\\a) gated by ``display.bell_on_prompt`` (``prompt=True``, blocking modals)
or ``display.bell_on_complete`` (end of turn); works over SSH. The same flag also emits the
OSC 9 / Warp OSC 777 desktop notification; ``context`` is the short notification body."""
flag = "bell_on_prompt" if prompt else "bell_on_complete"
if not getattr(self, flag, False) or getattr(self, "_terminal_io_broken", False):
return
from hermes_cli.cli_terminal_mixin import _run_on_app_loop, _write_terminal_sequence
from hermes_cli.terminal_notify import notification_sequence, write_tty
body = context or ("input needed" if prompt else "turn complete")
try:
seq = "\a" + notification_sequence(
body, prompt=prompt, session_id=getattr(self, "session_id", "") or "", detail=detail)
except Exception:
return
app = getattr(self, "_app", None)
if app is None or not getattr(app, "_is_running", False):
write_tty(seq)
return
# Agent thread. The loop thread may be mid-write of a 12 KB kitty pet frame that the tty
# drains ~1 KB at a time; a second writer on the same tty (/dev/tty, sys.stdout) splices
# in, the foreign ESC aborts the APC, and the terminal paints the rest of the payload as
# base64 at the input cursor. Serialize behind the renderer instead.
def _emit() -> None:
try:
_write_terminal_sequence(app, seq)
except (OSError, ValueError):
pass # dead tty: same fail-quiet as _pet_flush_kitty_frame
_run_on_app_loop(app, _emit)
def _clarify_teardown(self) -> None:
self._clarify_state = None
self._clarify_freetext = False
self._clarify_deadline = None
self._clarify_multi_base = None
self._paint_now()
def _clarify_callback(self, question, choices, multi_select=False, questions=None):
"""Clarify-tool platform callback (agent thread): show the selection UI (or freetext for
open-ended questions) and block until the key bindings answer or the timeout dismisses it
(the agent is then told to decide). ``multi_select`` shows checkboxes (Space toggles).
A non-empty ``questions`` list switches to the batch panel and returns
``{"answers": {qid: raw}}`` (plus ``"timed_out": True`` on a partial deadline expiry).
The single-question path below is unchanged. See #18450.
"""
from cli import CLI_CONFIG, _DIM, _RST, _cprint
from tools.clarify_gateway import resolve_clarify_timeout
if questions:
return self._clarify_callback_batch(questions)
# Canonical clarify timeout, shared with the gateway/TUI path; `<= 0` = unlimited.
timeout = resolve_clarify_timeout(CLI_CONFIG)
response_queue = queue.Queue()
is_open_ended = not choices
effective_multi = multi_select and not is_open_ended
self._clarify_state = {
"question": question,
"choices": choices if not is_open_ended else [],
"selected": 0,
"multi_select": effective_multi,
"selected_indices": set() if effective_multi else None,
"response_queue": response_queue}
self._clarify_deadline = None if timeout <= 0 else _time.monotonic() + timeout
self._clarify_freetext = is_open_ended # open-ended → straight to freetext
self._clarify_multi_base = None
self._ring_bell(prompt=True, context="clarify")
self._paint_now()
result = self._poll_modal_queue(response_queue, "_clarify_deadline")
if result is not _TIMED_OUT:
self._clarify_deadline = None
self._persist_prompt_summary("?", "Clarify", question, str(result))
return result
self._clarify_teardown()
_cprint(f"\n{_DIM}(clarify timed out after {timeout}s — agent will decide){_RST}")
return _CLARIFY_TIMEOUT_REPLY
# --- Connection setup --------------------------------------------------
def _connection_operation(self, payload):
from tools.connectors import live
session_key = getattr(self, "session_id", "") or ""
op_id = str(payload.get("op_id") or "")
return live.get(session_key, op_id) or live.current(session_key)
def _connection_install_hook(self) -> bool:
from tools.connectors.operation import ConnectionOperation
if ConnectionOperation.on_change is not None:
return False
ConnectionOperation.on_change = self._connection_on_change
return True
def _connection_restore_hook(self) -> None:
from tools.connectors.operation import ConnectionOperation
if ConnectionOperation.on_change == self._connection_on_change:
ConnectionOperation.on_change = None
def _connection_close(self) -> None:
self._connection_state = None
self._connection_restore_hook()
self._restore_modal_input_snapshot()
self._paint_now()
@staticmethod
def _connection_fields(target) -> list[dict]:
return [dict(field) for field in target.get("required_env") or () if isinstance(field, dict)]
@staticmethod
def _connection_opening_phase(target) -> str:
"""The phase a target opens in. A pending install or enable waits for the user's Connect
even with no fields to fill; a pending authorize is the backend still minting the link."""
target_state = target.get("state")
if target_state == "initiated" and target.get("connect_url"):
return "url"
if target_state == "failed":
return "failed"
if target_state == "pending" and target.get("action") != "authorize":
return "form"
return "waiting"
def _connection_show_target(self, payload, index: int) -> None:
targets = payload.get("targets") or []
if not targets:
self._connection_close()
return
index = max(0, min(index, len(targets) - 1))
target = targets[index]
fields = self._connection_fields(target)
previous = self._connection_state or {}
drafts = previous.setdefault("drafts", {})
target_draft = drafts.setdefault(target.get("name", ""), {})
for field in fields:
if "type" not in field:
field["type"] = "secret" if field.get("secret") else "plain"
if field.get("type") != "secret" and field.get("name") not in target_draft:
target_draft[field.get("name")] = str(field.get("default") or "")
self._connection_state = {
**previous,
"payload": payload,
"target_index": index,
"target": target,
"fields": fields,
"field_index": 0,
"selected": 0,
"phase": self._connection_opening_phase(target),
"drafts": drafts,
}
self._connection_sync_input_buffer()
self._paint_now()
def _connection_active_field_is_secret(self) -> bool:
state = self._connection_state
if not state or state.get("phase") not in {"form", "failed"}:
return False
fields = state.get("fields") or []
index = state.get("field_index", 0)
return 0 <= index < len(fields) and fields[index].get("type") == "secret"
def _connection_prefill_text(self) -> str:
state = self._connection_state
if not state or self._connection_active_field_is_secret():
return ""
fields = state.get("fields") or []
index = state.get("field_index", 0)
if not (0 <= index < len(fields)):
return ""
field = fields[index]
target_name = state.get("target", {}).get("name", "")
return str(state.get("drafts", {}).get(target_name, {}).get(field.get("name"), ""))
def _connection_sync_input_buffer(self) -> None:
app = getattr(self, "_app", None)
if app is None:
return
def _apply() -> None:
try:
buf = app.current_buffer
buf.text = self._connection_prefill_text()
buf.cursor_position = len(buf.text)
except Exception:
pass
loop = getattr(app, "loop", None)
if loop is not None and threading.current_thread() is not threading.main_thread():
try:
loop.call_soon_threadsafe(_apply)
return
except Exception:
pass
_apply()
def _connection_callback(self, payload):
if not isinstance(payload, dict):
return None
self._capture_modal_input_snapshot()
installed = self._connection_install_hook()
self._connection_show_target(payload, 0)
state = self._connection_state
if state is None:
return None
state["owns_hook"] = installed
state["tool_thread_id"] = threading.current_thread().ident
if state["phase"] != "waiting":
self._ring_bell(prompt=True, context="connection setup")
return None
def _connection_answer(self, *, approve: bool) -> None:
state = self._connection_state
if not state:
return
target = state["target"]
operation = self._connection_operation(state["payload"])
if operation is None:
self._connection_close()
return
name = str(target.get("name") or "")
answer = {"name": name, "status": "approved" if approve else "skipped"}
if approve:
answer["env"] = dict(state.get("drafts", {}).get(name, {}))
from tools.connectors.mcp import apply_answer
# The backend applies the answer on this thread, and its change hook sets the next phase
# (the URL step, the form again for a missing field) before apply_answer returns. Set the
# waiting phase first so it cannot overwrite that.
state["phase"] = "waiting"
apply_answer(operation, json.dumps({"targets": [answer]}))
self._paint_now()
def _connection_retry(self) -> None:
state = self._connection_state
if not state:
return
operation = self._connection_operation(state["payload"])
if operation is None:
self._connection_close()
return
target = state["target"]
if target.get("kind") == "connector" and target.get("state") in {"failed", "expired"}:
from tools.connectors.run import reissue
state["phase"] = "waiting"
if reissue(operation, [str(target.get("name") or "")]) is not None:
target["detail"] = "This connection cannot be started again. Cancel and ask the agent again."
state["phase"] = "failed"
self._paint_now()
return
if target.get("state") in {"pending", "failed", "expired"}:
# Connect on a failed row is the same attempt with the values now in the draft.
self._connection_answer(approve=True)
return
from tools.connectors.mcp import retry
state["phase"] = "waiting"
retry(operation, [str(target.get("name") or "")])
self._paint_now()
def _connection_continue(self) -> None:
state = self._connection_state
if not state:
return
operation = self._connection_operation(state["payload"])
if operation is not None:
from tools.connectors.mcp import apply_answer
apply_answer(operation, json.dumps({"settled_by": "continue"}))
self._connection_close()
def _connection_open_url(self) -> None:
state = self._connection_state
if state and state.get("phase") == "url" and state["target"].get("connect_url"):
webbrowser.open(state["target"]["connect_url"])
def _connection_cancel(self) -> None:
self._connection_answer(approve=False)
def _connection_interrupt(self) -> None:
state = self._connection_state
if state:
from tools.interrupt import set_interrupt
operation = self._connection_operation(state["payload"])
set_interrupt(True, thread_id=state.get("tool_thread_id"))
if operation is not None:
operation.wake.set()
self._connection_close()
def _connection_on_change(self, operation, _change, snapshot) -> None:
state = self._connection_state
if not state or operation.op_id != state["payload"].get("op_id"):
return
if snapshot.get("settled_at") is not None:
self._connection_close()
return
active_name = state["target"].get("name")
target = next((item for item in snapshot.get("targets") or () if item.get("name") == active_name), None)
if target is None:
return
state["payload"] = snapshot
state["target"] = target
target_state = target.get("state")
if target_state == "initiated" and target.get("connect_url"):
state["phase"] = "url"
elif target_state == "failed":
state["phase"] = "failed"
elif target_state == "pending" and self._connection_fields(target):
# The backend refused the answer because a required field is still empty: reopen the
# form on the first one it named, over the draft the panel kept.
missing = {field.get("name") for field in self._connection_fields(target)}
names = [field.get("name") for field in state.get("fields") or []]
state["field_index"] = next((i for i, name in enumerate(names) if name in missing), 0)
state["phase"] = "form"
self._connection_sync_input_buffer()
elif target_state == "connected" and target.get("discovery_error"):
state["phase"] = "authorized"
elif target_state in {"connected", "skipped"}:
unresolved = [
item for item in snapshot.get("targets") or ()
if item.get("state") not in {"connected", "skipped"}
]
if unresolved:
self._connection_show_target(snapshot, (snapshot.get("targets") or []).index(unresolved[0]))
return
state["phase"] = "connected"
else:
state["phase"] = "waiting"
self._paint_now()
def _connection_set_field(self, value: str) -> None:
"""Save the active input row in the private draft and advance to the next row/action."""
state = self._connection_state
if not state or state.get("phase") not in {"form", "failed"}:
return
fields = state.get("fields") or []
index = state.get("field_index", 0)
if not (0 <= index < len(fields)):
return
field = fields[index]
state["drafts"][state["target"].get("name", "")][field.get("name")] = value
state["field_index"] = min(index + 1, len(fields))
self._connection_sync_input_buffer()
self._paint_now()
def _connection_submit(self) -> None:
"""Enter action for prompt_toolkit bindings: URL open, selected action, or field advance."""
state = self._connection_state
if not state:
return
phase = state.get("phase")
if phase == "url":
self._connection_open_url()
elif phase == "authorized":
self._connection_continue()
elif phase in {"form", "failed"} and state.get("field_index", 0) >= len(state.get("fields") or []):
(self._connection_retry if state.get("selected", 0) == 0 else self._connection_cancel)()
def _connection_field_lines(self, state, target) -> list[str]:
draft = state.get("drafts", {}).get(target.get("name", ""), {})
lines = []
for field in state.get("fields") or ():
marker = "*" if field.get("required") else ""
value = "Set" if field.get("type") == "secret" and draft.get(field.get("name")) else draft.get(field.get("name"), "")
lines.append(f"{field.get('prompt') or field.get('name')}{marker}: {value}")
return lines
def _connection_render_lines(self) -> list[str]:
"""Panel copy without exposing secret drafts; the prompt_toolkit renderer consumes these rows."""
state = self._connection_state
if not state:
return []
target = state["target"]
phase = state.get("phase")
lines = [f"Set up {target.get('name', '')}"]
if target.get("instructions"):
lines.append(str(target["instructions"]))
if phase == "form":
lines.extend(self._connection_field_lines(state, target))
lines.append("Connect Cancel")
elif phase == "url":
lines.extend([str(target.get("connect_url") or ""), str(target.get("detail") or ""), "Press Enter to open in browser"])
elif phase == "failed":
lines.append(str(target.get("detail") or "Connection failed"))
lines.extend(self._connection_field_lines(state, target))
lines.append("Connect Cancel")
elif phase == "authorized":
# A connected row cannot be re-run inside this operation; the agent retries discovery
# with its next manage_connections call, which needs no new consent.
lines.extend(["Authorized. Tools unavailable.", "Continue"])
elif phase == "connected":
lines.append("Connected")
else:
lines.append(str(target.get("detail") or "Waiting…"))
return [line for line in lines if line]
# --- Batch clarify (multi-question, issue #18450) -----------------------
def _clarify_batch_set_active(self, state, index) -> None:
"""Point the batch clarify panel at question ``index``: mirror it into the flat keys the
single-question keybindings/renderer read so ↑/↓/Space/number keys work unchanged;
open-ended drops into freetext; re-visiting restores the earlier cursor/checkboxes."""
questions_list = state["questions"]
index = max(0, min(index, len(questions_list) - 1))
entry = questions_list[index]
choices = entry["choices"] or []
state["active"] = index
state["question"] = entry["question"]
state["choices"] = choices
state["selected"] = 0
state["multi_select"] = bool(entry["multi_select"])
state["selected_indices"] = set() if entry["multi_select"] else None
self._clarify_freetext = not entry["choices"]
self._clarify_multi_base = None
meta = (state.get("answer_meta") or {}).get(entry["qid"])
if meta is None:
return
kind = meta.get("kind")
if kind == "choice":
answer = state["answers"].get(entry["qid"])
if answer in choices:
state["selected"] = choices.index(answer)
elif kind == "other":
state["selected"] = len(choices)
elif kind == "multi":
checked = {choices.index(c) for c in meta.get("choices") or [] if c in choices}
if meta.get("other_text"):
checked.add(len(choices))
state["selected_indices"] = checked
def _clarify_batch_lock(self, state, answer, meta=None) -> None:
"""Lock ``answer`` for the active batch question (overwriting an earlier one) and advance to
the next unanswered; ``meta`` ({"kind": "choice"|"other"|"multi", ...}) lets a re-visit
restore the cursor / prefill an "Other" edit. All answered → resolve the queue, tear down."""
entry = state["questions"][state["active"]]
state["answers"][entry["qid"]] = answer
state.setdefault("answer_meta", {})[entry["qid"]] = meta or {"kind": "choice"}
self._persist_prompt_summary("?", "Clarify", entry["question"], str(answer))
total = len(state["questions"])
for offset in range(1, total + 1):
candidate = (state["active"] + offset) % total
if state["questions"][candidate]["qid"] not in state["answers"]:
self._clarify_batch_set_active(state, candidate)
return
try:
state["response_queue"].put(dict(state["answers"]))
except Exception:
pass
self._clarify_state = None
self._clarify_freetext = False
self._clarify_multi_base = None
def _clarify_batch_enter(self, state) -> None:
"""Enter in batch choice mode: lock the active selection. Multi-select locks a JSON array of
checked labels (parsed by the tool core); "Other" switches to freetext, prefilled with an
earlier typed answer so Enter on an answered Other edits instead of retyping."""
choices = state.get("choices") or []
selected = state.get("selected", 0)
entry = state["questions"][state["active"]]
meta = (state.get("answer_meta") or {}).get(entry["qid"]) or {}
if state.get("multi_select"):
sorted_idx = sorted(state.get("selected_indices") or set())
selected_choices = [choices[i] for i in sorted_idx if i < len(choices)]
if len(choices) in sorted_idx:
# Stash the checked real choices so the freetext submit appends the typed answer.
self._clarify_multi_base = selected_choices
self._clarify_freetext = True
self._clarify_prefill = meta.get("other_text") or ""
return
self._clarify_batch_lock(
state,
json.dumps(selected_choices, ensure_ascii=False),
meta={"kind": "multi", "choices": selected_choices, "other_text": ""})
return
if selected < len(choices):
self._clarify_batch_lock(state, choices[selected], meta={"kind": "choice"})
return
self._clarify_freetext = True
self._clarify_prefill = meta.get("other_text") or "" if meta.get("kind") == "other" else ""
def _clarify_callback_batch(self, questions):
"""Batch clarify panel (A-compact): all questions, one active. Returns
``{"answers": {qid: raw}}`` when every question is locked, plus ``"timed_out": True`` when
the deadline expires with partial answers; a cancel string passes through unchanged so the
tool core resolves the batch empty."""
from cli import CLI_CONFIG, _DIM, _RST, _cprint
from tools.clarify_gateway import resolve_clarify_timeout
timeout = resolve_clarify_timeout(CLI_CONFIG)
response_queue = queue.Queue()
state = {
"questions": list(questions),
"answers": {},
"answer_meta": {},
"active": 0,
"response_queue": response_queue,
# Flat keys mirroring the active question — filled by _clarify_batch_set_active.
"question": "",
"choices": [],
"selected": 0,
"multi_select": False,
"selected_indices": None}
self._clarify_state = state
self._clarify_batch_set_active(state, 0)
self._clarify_deadline = None if timeout <= 0 else _time.monotonic() + timeout
self._ring_bell(prompt=True, context="clarify")
self._paint_now()
result = self._poll_modal_queue(response_queue, "_clarify_deadline")
if result is not _TIMED_OUT:
self._clarify_deadline = None
return {"answers": result} if isinstance(result, dict) else result
partial = dict(state["answers"])
self._clarify_teardown()
_cprint(f"\n{_DIM}(clarify timed out after {timeout}s — locked answers returned){_RST}")
return {"answers": partial, "timed_out": True}
def _sudo_password_callback(self) -> str:
"""Prompt for a sudo password through the prompt_toolkit UI (agent thread); clarify-style
state + queue answered by the Enter binding."""
from cli import _DIM, _RST, _cprint
response_queue = queue.Queue()
self._capture_modal_input_snapshot()
self._sudo_state = {"response_queue": response_queue}
self._sudo_deadline = _time.monotonic() + 45
self._ring_bell(prompt=True, context="sudo password")
self._paint_now()
result = self._poll_modal_queue(response_queue, "_sudo_deadline", refresh=0)
self._sudo_state = None
self._sudo_deadline = 0
self._restore_modal_input_snapshot()
self._paint_now()
if result is _TIMED_OUT:
_cprint(f"\n{_DIM} ⏱ Timeout — continuing without sudo{_RST}")
return ""
if result:
_cprint(f"\n{_DIM} ✓ Password received (cached for session){_RST}")
else:
_cprint(f"\n{_DIM} ⏭ Skipped{_RST}")
return result
def _approval_callback(self, command: str, description: str,
*, allow_permanent: bool = True,
allow_session: bool = True,
smart_denied: bool = False) -> str:
"""Dangerous-command approval through the prompt_toolkit UI (agent thread).
Choices: once / session / always / deny (see ``_approval_choices``), plus 'view' for long
commands. ``_approval_lock`` serializes concurrent requests (parallel delegation subtasks)
so the shared ``_approval_state`` / ``_approval_deadline`` aren't clobbered.
"""
from cli import CLI_CONFIG, _DIM, _RST, _cprint
with self._approval_lock:
timeout = int(CLI_CONFIG.get("approvals", {}).get("timeout", 300))
response_queue = queue.Queue()
self._approval_state = {
"command": command,
"description": description,
"choices": self._approval_choices(
command,
allow_permanent=allow_permanent,
allow_session=allow_session,
smart_denied=smart_denied),
"selected": 0,
"response_queue": response_queue}
self._approval_deadline = _time.monotonic() + timeout
self._ring_bell(prompt=True, context="approval", detail=command)
self._paint_now()
result = self._poll_modal_queue(response_queue, "_approval_deadline")
self._approval_state = None
self._approval_deadline = 0
self._paint_now()
if result is _TIMED_OUT:
_cprint(f"\n{_DIM} ⏱ Timeout — denying command{_RST}")
self._persist_prompt_summary("⚠", "Approval", command, "timed out (no response)")
return "timeout"
self._persist_prompt_summary(
"⚠", "Approval", command, _APPROVAL_OUTCOME_LABELS.get(result, str(result)))
return result
def _approval_choices(self, command: str, *, allow_permanent: bool = True,
allow_session: bool = True,
smart_denied: bool = False) -> list[str]:
"""Smart-DENY overrides and re-ask-every-time gates (allow_session=False) show only
once/deny; ``allow_permanent=False`` for another reason (e.g. tirith) hides only 'always'."""
if smart_denied or not allow_session:
choices = ["once", "deny"]
elif allow_permanent:
choices = ["once", "session", "always", "deny"]
else:
choices = ["once", "session", "deny"]
if len(command) > 70:
choices.append("view")
return choices
def _handle_approval_selection(self) -> None:
"""Process the currently selected dangerous-command approval choice."""
state = self._approval_state
if not state:
return
selected = state.get("selected", 0)
choices = state.get("choices")
if not isinstance(choices, list):
choices = []
if not (0 <= selected < len(choices)):
return
chosen = choices[selected]
if chosen == "view":
state["show_full"] = True
state["choices"] = [choice for choice in choices if choice != "view"]
if state["selected"] >= len(state["choices"]):
state["selected"] = max(0, len(state["choices"]) - 1)
self._invalidate()
return
state["response_queue"].put(chosen)
self._approval_state = None
self._invalidate()
def _vault_unlock_callback(self, backend_name: str, display_name: str) -> str:
"""Masked master-password prompt for an external password manager (agent thread).
Reuses the sudo panel state so rendering, Enter/ESC handling and interrupt cleanup are shared."""
from cli import _DIM, _RST, _cprint
response_queue = queue.Queue()
self._capture_modal_input_snapshot()
self._sudo_state = {"response_queue": response_queue, "vault_backend": display_name}
self._sudo_deadline = _time.monotonic() + 120
self._ring_bell(prompt=True, context=f"unlock {display_name}")
self._paint_now()
result = self._poll_modal_queue(response_queue, "_sudo_deadline", refresh=0)
self._sudo_state = None
self._sudo_deadline = 0
self._restore_modal_input_snapshot()
self._paint_now()
if result is _TIMED_OUT or not result:
_cprint(f"\n{_DIM} ⏭ {display_name} stays locked{_RST}")
return ""
_cprint(f"\n{_DIM} ✓ Unlocking {display_name} for this session{_RST}")
return result
def _vault_save_login_callback(self, origin: str, site: str):
"""Two-step "save this login" prompt (identifier shown, password masked) on the sudo panel; the
answer goes to the vault store, never to the model. None = declined."""
from cli import _DIM, _RST, _cprint
answer: dict = {}
for step in ("identifier", "password"):
response_queue = queue.Queue()
self._capture_modal_input_snapshot()
self._sudo_state = {"response_queue": response_queue, "vault_save": {"site": site, "origin": origin,
"step": step}}
self._sudo_deadline = _time.monotonic() + 180
if step == "identifier":
self._ring_bell(prompt=True, context=f"save login for {site}")
self._paint_now()
result = self._poll_modal_queue(response_queue, "_sudo_deadline", refresh=0)
self._sudo_state = None
self._sudo_deadline = 0
self._restore_modal_input_snapshot()
self._paint_now()
if result is _TIMED_OUT or not result:
_cprint(f"\n{_DIM} ⏭ Not saving a login for {site}{_RST}")
return None
answer[step] = result
_cprint(f"\n{_DIM} ✓ Login for {site} saved to your vault{_RST}")
return answer
def _vault_code_callback(self, site: str, hint: str) -> str:
"""One-time-code prompt (shown as typed; a 6-digit code is not a secret worth masking and users
need to see typos) on the sudo panel. "" = declined/timed out."""
from cli import _DIM, _RST, _cprint
response_queue = queue.Queue()
self._capture_modal_input_snapshot()
self._sudo_state = {"response_queue": response_queue, "vault_code": {"site": site, "hint": hint}}
self._sudo_deadline = _time.monotonic() + 180
self._ring_bell(prompt=True, context=f"verification code for {site}")
self._paint_now()
result = self._poll_modal_queue(response_queue, "_sudo_deadline", refresh=0)
self._sudo_state = None
self._sudo_deadline = 0
self._restore_modal_input_snapshot()
self._paint_now()
if result is _TIMED_OUT or not result:
_cprint(f"\n{_DIM} ⏭ No code entered for {site}{_RST}")
return ""
_cprint(f"\n{_DIM} ✓ Code entered into {site}{_RST}")
return result
def _secret_capture_callback(self, var_name: str, prompt: str, metadata=None) -> dict:
self._capture_modal_input_snapshot()
try:
return prompt_for_secret(self, var_name, prompt, metadata)
finally:
self._restore_modal_input_snapshot()
self._paint_now()
def _capture_modal_input_snapshot(self) -> None:
"""Temporarily clear the input buffer and save the user's in-progress draft."""
if getattr(self, "_modal_input_snapshot", None) is not None or not getattr(self, "_app", None):
return
try:
buf = self._app.current_buffer
self._modal_input_snapshot = {"text": buf.text, "cursor_position": buf.cursor_position}
buf.reset()
except Exception:
self._modal_input_snapshot = None
def _restore_modal_input_snapshot(self) -> None:
"""Restore any draft text that was present before a modal prompt opened."""
snapshot = getattr(self, "_modal_input_snapshot", None)
self._modal_input_snapshot = None
if not snapshot or not getattr(self, "_app", None):
return
try:
buf = self._app.current_buffer
buf.text = snapshot.get("text", "")
buf.cursor_position = min(snapshot.get("cursor_position", 0), len(buf.text))
except Exception:
pass
def _clear_active_overlays_for_interrupt(self) -> None:
"""Drain and clear every input-blocking overlay left by an interrupted agent: the worker
thread is gone but the state dict still gates input (frozen terminal until its timeout).
Push a safe value onto each queue (approval -> "deny", others -> cancel), nil the state,
restore the draft; each step is wrapped so a dead queue can't block the others."""
def _put(state, value) -> None:
try:
state["response_queue"].put(value)
except Exception:
pass
if self._approval_state:
_put(self._approval_state, "deny")
self._approval_state = None
if self._connection_state:
self._connection_interrupt()
if self._clarify_state:
_put(self._clarify_state, "The user cancelled. Use your best judgement to proceed.")
self._clarify_state = None
self._clarify_freetext = False
self._clarify_multi_base = None
if self._sudo_state:
_put(self._sudo_state, "")
self._sudo_state = None
self._sudo_deadline = 0
self._restore_modal_input_snapshot()
if self._secret_state:
try:
self._cancel_secret_capture()
except Exception:
self._secret_state = None
def _submit_secret_response(self, value: str) -> None:
if not self._secret_state:
return
self._secret_state["response_queue"].put(value)
self._secret_state = None
self._secret_deadline = 0
self._paint_now() # direct paint so the secret panel clears at once (no throttle)
def _cancel_secret_capture(self) -> None:
self._submit_secret_response("")
def _clear_secret_input_buffer(self) -> None:
if getattr(self, "_app", None):
try:
self._app.current_buffer.reset()
except Exception:
pass