hermes-admin/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 28 Packages Projects Releases Wiki Activity

merge origin/main (779 commits) into ethie/pm-clean

Browse Source 
Branch semantics kept where main and PM disagree: update_cmd_deps.py,
constraints-termux.txt, the Electron update-api-check module and the
post-swap hand-off test stay deleted; the pending-fleet-restart catch-up
and the local_runtime tag/download ladder stay retired (PM owns engines).

Ported from main onto the branch's shape: profile_scoped_chore for the
auto-archive and plugin-update housekeeping chores, the local-runtime
cross-process boot lock and residency cap, the checkpoint tmp_pack sweep,
the cua daemon-liveness status probe, the remote-served Desktop update
flag (posix.sh / windows.ps1), sign-in for env-pinned remote gateways
(urlDisabled on RemoteSetupFields), the uvloop extra split (uvicorn
without [standard]), and the umask-scoping spawn test.

uv.lock regenerated with pm.build_env --lock-only; new utf-8 reads from
main switched to utf-8-sig (check-windows-footguns).
This commit is contained in:
ethernet
2026-09-21 00:58:39 -04:00
parent 3b181dd848 afc3b7c6f3
commit 9f2ba1b74d
1384 changed files with 51561 additions and 26427 deletions
Download Patch File Download Diff File Expand all files Collapse all files

20
acp_adapter/commands.py
View File

@@ -36,6 +36,15 @@ def _queue_prompt(state: SessionState, text: str) -> int:
return len(state.queued_prompts)
# Commands that mutate shared turn state must not run beside a live turn or beside each
# other: slash dispatch happens on a worker thread while the turn iterates state.history and
# reads agent._session_db, so clearing, rebinding, or swapping state.agent underneath
# run_conversation tears the running turn. Gateway parity: all three are idle-only there.
# The command_op flag is held for the whole handler so a turn cannot claim the session in
# the check-then-act window (the /compress LLM call and /model agent rebuild take seconds).
_MID_TURN_BLOCKED_COMMANDS = frozenset({"reset", "compress", "model"})
class SlashCommandsMixin:
"""Slash-command surface for ``HermesACPAgent``; relies on ``_conn``, ``_send``, ``_schedule_soon``,
``session_manager`` and ``_switch_model`` from the host class."""
@@ -93,6 +102,13 @@ class SlashCommandsMixin:
if cmd not in self._COMMANDS:
return None
mutating = cmd in _MID_TURN_BLOCKED_COMMANDS
if mutating:
with state.runtime_lock:
if state.is_running or state.command_op:
return (f"⏳ Session is busy; /{cmd} only works while the session is "
"idle. Wait for the current response or cancel first.")
state.command_op = True
handler = getattr(self, f"_cmd_{cmd}")
# Handlers run outside the per-turn cwd-pinning context. ``/compress``
@@ -112,6 +128,10 @@ class SlashCommandsMixin:
except Exception as e:
logger.error("Slash command /%s error: %s", cmd, e, exc_info=True)
return f"Error executing /{cmd}: {e}"
finally:
if mutating:
with state.runtime_lock:
state.command_op = False
def _cmd_help(self, args: str, state: SessionState) -> str:
lines = ["Available commands:", ""]

19
acp_adapter/edit_approval.py
View File

@@ -30,6 +30,10 @@ class EditProposal:
old_text: str | None
new_text: str
arguments: dict[str, Any]
# Every file the edit will actually touch. ``path`` may be a comma-joined
# display string for multi-file V4A patches; ``paths`` is the authoritative
# set for auto-approve checks. Empty means "``path`` alone".
paths: tuple[str, ...] = ()
EditApprovalRequester = Callable[[EditProposal], bool]
@@ -118,6 +122,7 @@ def _proposal_for_patch_v4a(arguments: dict[str, Any]) -> EditProposal:
return EditProposal(
"patch", paths[0] if single else ", ".join(paths),
_read_text_if_exists(paths[0]) if single else None, patch_body, dict(arguments),
tuple(paths),
)
@@ -145,16 +150,22 @@ def should_auto_approve_edit(proposal: EditProposal, policy: str, cwd: str | Non
Session-scoped and conservative: sensitive paths still ask under autonomous policies."""
policy = str(policy or AUTO_APPROVE_ASK).strip()
if policy == AUTO_APPROVE_ASK or _is_sensitive_auto_approve_path(proposal.path):
# Multi-file V4A proposals join paths into one display string; the checks
# must run per real target or a sensitive/escaped file hides in the join.
paths = proposal.paths or (proposal.path,)
if policy == AUTO_APPROVE_ASK or any(_is_sensitive_auto_approve_path(p) for p in paths):
return False
path = Path(proposal.path).expanduser().resolve(strict=False)
resolved = [Path(p).expanduser().resolve(strict=False) for p in paths]
if policy == AUTO_APPROVE_SESSION:
return True
if policy == AUTO_APPROVE_WORKSPACE:
# tempfile.gettempdir() is the real temp root on every platform
# (``/private/tmp`` on macOS since resolve() follows the symlink).
return path.is_relative_to(Path(tempfile.gettempdir()).resolve(strict=False)) or (
bool(cwd) and path.is_relative_to(Path(cwd).expanduser().resolve(strict=False)))
tmp = Path(tempfile.gettempdir()).resolve(strict=False)
ws = Path(cwd).expanduser().resolve(strict=False) if cwd else None
return all(
path.is_relative_to(tmp) or (ws is not None and path.is_relative_to(ws))
for path in resolved)
return False

131
acp_adapter/server.py
View File

@@ -728,11 +728,12 @@ class HermesACPAgent(SlashCommandsMixin, acp.Agent):
"""Mark the session running; if a turn is active, redirect it (text-only, supported
runtime) or queue it. Returns the client message when absorbed, else None."""
with state.runtime_lock:
if not state.is_running:
if not state.is_running and not state.command_op:
state.is_running = True
state.current_prompt_text = user_text or "[Image attachment]"
return None
if text_only and isinstance(user_content, str) and hasattr(state.agent, "redirect") and (
# Redirect steers a live turn; a state-mutating command (command_op) has none.
if state.is_running and text_only and isinstance(user_content, str) and hasattr(state.agent, "redirect") and (
getattr(state.agent, "_supports_active_turn_redirect", False) is True
):
try:
@@ -831,6 +832,8 @@ class HermesACPAgent(SlashCommandsMixin, acp.Agent):
if self._conn:
await self._conn.session_update(session_id, acp.update_agent_message_text(response_text))
await self._send_usage_update(state)
# A mutating command held command_op; prompts that arrived mid-op are queued.
await self._drain_queued_prompts(state, session_id, self._conn)
return PromptResponse(stop_reason="end_turn")
absorbed = self._claim_turn_or_queue(state, session_id, user_text, user_content, text_only_prompt)
@@ -940,55 +943,50 @@ class HermesACPAgent(SlashCommandsMixin, acp.Agent):
streamed_message: bool,
) -> PromptResponse:
"""Persist, emit provenance/final text, drain queued prompts, report usage."""
# Key presence, not truthiness: ``messages=[]`` is a legitimate cleared transcript (#10844);
# only a result without the key leaves the history untouched.
if "messages" in result and isinstance(result["messages"], list):
state.history = result["messages"]
self.session_manager.save_session(session_id)
try:
# Key presence, not truthiness: ``messages=[]`` is a legitimate cleared transcript (#10844);
# only a result without the key leaves the history untouched.
if "messages" in result and isinstance(result["messages"], list):
state.history = result["messages"]
self.session_manager.save_session(session_id)
# Head rotated (compression split): emit provenance so clients can render the boundary.
post_turn_hermes_id = getattr(state.agent, "session_id", None)
if conn and post_turn_hermes_id and pre_turn_hermes_id and post_turn_hermes_id != pre_turn_hermes_id:
try:
await self._send_session_info_update(
session_id, current_hermes_session_id=post_turn_hermes_id,
previous_hermes_session_id=pre_turn_hermes_id,
)
except Exception:
logger.debug("Could not emit ACP provenance update after rotation for %s", session_id, exc_info=True)
# Head rotated (compression split): emit provenance so clients can render the boundary.
post_turn_hermes_id = getattr(state.agent, "session_id", None)
if conn and post_turn_hermes_id and pre_turn_hermes_id and post_turn_hermes_id != pre_turn_hermes_id:
try:
await self._send_session_info_update(
session_id, current_hermes_session_id=post_turn_hermes_id,
previous_hermes_session_id=pre_turn_hermes_id,
)
except Exception:
logger.debug("Could not emit ACP provenance update after rotation for %s", session_id, exc_info=True)
final_response = result.get("final_response") or "" # None on an interrupted turn
cancelled = bool(state.cancel_event and state.cancel_event.is_set())
# The local "waiting for model" interrupt status is metadata, not prose; stop_reason carries it.
from agent.conversation_loop import INTERRUPT_WAITING_FOR_MODEL_PREFIX
final_response = result.get("final_response") or "" # None on an interrupted turn
cancelled = bool(state.cancel_event and state.cancel_event.is_set())
# The local "waiting for model" interrupt status is metadata, not prose; stop_reason carries it.
from agent.conversation_loop import INTERRUPT_WAITING_FOR_MODEL_PREFIX
interrupted = bool(result.get("interrupted")) or cancelled
suppress = interrupted and final_response.startswith(INTERRUPT_WAITING_FOR_MODEL_PREFIX)
# Send the final text unless already streamed — or if a plugin hook transformed it after.
if final_response and conn and not suppress and (not streamed_message or result.get("response_transformed")):
update = acp.update_agent_message_text(final_response)
if state.message_ids is not None:
# A plugin-rewritten reply replaces the streamed bubble (same id); an
# unstreamed final response opens its own.
if streamed_message and result.get("response_transformed"):
update.message_id = state.message_ids.last() or state.message_ids.current()
else:
update.message_id = state.message_ids.current()
state.message_ids.close()
await conn.session_update(session_id, update)
interrupted = bool(result.get("interrupted")) or cancelled
suppress = interrupted and final_response.startswith(INTERRUPT_WAITING_FOR_MODEL_PREFIX)
# Send the final text unless already streamed — or if a plugin hook transformed it after.
if final_response and conn and not suppress and (not streamed_message or result.get("response_transformed")):
update = acp.update_agent_message_text(final_response)
if state.message_ids is not None:
# A plugin-rewritten reply replaces the streamed bubble (same id); an
# unstreamed final response opens its own.
if streamed_message and result.get("response_transformed"):
update.message_id = state.message_ids.last() or state.message_ids.current()
else:
update.message_id = state.message_ids.current()
state.message_ids.close()
await conn.session_update(session_id, update)
# Go idle before draining so recursive prompt() calls can acquire the session.
with state.runtime_lock:
state.is_running = False
state.current_prompt_text = ""
while True:
finally:
# Go idle before draining so recursive prompt() calls can acquire the session.
with state.runtime_lock:
if not state.queued_prompts:
break
next_prompt = state.queued_prompts.pop(0)
if conn:
await conn.session_update(session_id, acp.update_user_message_text(next_prompt))
await self.prompt(prompt=[TextContentBlock(type="text", text=next_prompt)], session_id=session_id)
state.is_running = False
state.current_prompt_text = ""
await self._drain_queued_prompts(state, session_id, conn)
usage = None
if any(result.get(k) is not None for k in ("prompt_tokens", "completion_tokens", "total_tokens")):
@@ -1000,12 +998,34 @@ class HermesACPAgent(SlashCommandsMixin, acp.Agent):
await self._send_usage_update(state)
return PromptResponse(stop_reason="cancelled" if cancelled else "end_turn", usage=usage)
async def _drain_queued_prompts(self, state: SessionState, session_id: str, conn: Any) -> None:
"""Run queued prompts while the session is idle. Reached from ``_finish_turn`` and
from the slash path after a state-mutating command releases ``command_op``."""
while True:
with state.runtime_lock:
if state.is_running or state.command_op or not state.queued_prompts:
return
next_prompt = state.queued_prompts.pop(0)
if conn:
await conn.session_update(session_id, acp.update_user_message_text(next_prompt))
await self.prompt(prompt=[TextContentBlock(type="text", text=next_prompt)], session_id=session_id)
# ---- Session settings (ACP protocol methods) -----------------------------
async def set_session_model(self, model_id: str, session_id: str, **kwargs: Any) -> SetSessionModelResponse | None:
"""Switch the model for a session (called by ACP protocol)."""
state = await asyncio.to_thread(self.session_manager.get_session, session_id)
if state:
if state is None:
logger.warning("Session %s: model switch requested for missing session", session_id)
return None
# The picker swaps state.agent wholesale; mid-turn that strands the running agent and
# makes _finish_turn emit a spurious compression-rotation update. Same exclusion as
# the /model slash command.
with state.runtime_lock:
if state.is_running or state.command_op:
raise acp.RequestError(-32603, "Session is busy; switch models while the session is idle")
state.command_op = True
try:
# switch_model() does synchronous network I/O (models.dev, custom-endpoint probes,
# ~10 s cold) — off the loop, like the gateway, so other ACP sessions keep flowing.
try:
@@ -1018,12 +1038,17 @@ class HermesACPAgent(SlashCommandsMixin, acp.Agent):
# (disabled provider, context window below the floor) stays on the -32603 path.
from acp.exceptions import RequestError
raise RequestError.invalid_params({"details": str(exc)}) from exc
logger.info(
"Session %s: model switched to %s via provider %s", session_id, resolved_model, requested_provider
)
return SetSessionModelResponse()
logger.warning("Session %s: model switch requested for missing session", session_id)
return None
finally:
with state.runtime_lock:
state.command_op = False
# Drain AFTER this response is queued, never inside it: a prompt that arrived
# mid-switch would otherwise run a whole turn before the client sees the
# (possibly failed) switch result.
self._schedule_soon(lambda: self._drain_queued_prompts(state, session_id, self._conn))
logger.info(
"Session %s: model switched to %s via provider %s", session_id, resolved_model, requested_provider
)
return SetSessionModelResponse()
async def set_session_mode(self, mode_id: str, session_id: str, **kwargs: Any) -> SetSessionModeResponse | None:
"""Persist the editor-requested mode so ACP clients do not fail on mode switches."""

76
acp_adapter/session.py
View File

@@ -139,6 +139,11 @@ class SessionState:
history: List[Dict[str, Any]] = field(default_factory=list)
cancel_event: Any = None # threading.Event
is_running: bool = False
# A state-mutating slash command (/reset, /compress, /model) is in flight. Turn claims
# must queue behind it: /compress's LLM call and /model's agent rebuild take seconds, so
# a bare is_running check in the slash thread would leave a check-then-act window where
# a prompt claims the turn mid-mutation.
command_op: bool = False
queued_prompts: List[str] = field(default_factory=list)
runtime_lock: Any = field(default_factory=threading.Lock)
current_prompt_text: str = ""
@@ -164,6 +169,7 @@ class SessionManager:
self._restore_lock = threading.Lock()
self._agent_factory = agent_factory
self._db_instance = db # None → lazy-init on first use
self._cwd_backfilled = False
# ---- public API ---------------------------------------------------------
@@ -252,6 +258,11 @@ class SessionManager:
state.cwd = cwd
_register_task_cwd(session_id, cwd)
self._persist(state)
# Promote the authoritative column and claim an ordering generation, the
# same contract tui_gateway/session_workdir.py uses: a git probe may only
# publish while its generation is still current, so a slow probe for a
# previous workspace cannot overwrite a newer claim (A -> B -> A).
self._schedule_git_metadata(state, self._claim_cwd_generation(state))
return state
def save_session(self, session_id: str) -> None:
@@ -288,6 +299,16 @@ class SessionManager:
self._db_instance = acquire(get_hermes_home() / "state.db")
except Exception:
logger.debug("SessionDB unavailable for ACP persistence", exc_info=True)
if self._db_instance is not None and not self._cwd_backfilled:
# Rows minted before the adapter wrote the cwd column still carry the workspace
# in model_config; one idempotent UPDATE per process repairs them (#115705).
self._cwd_backfilled = True
try:
repaired = self._db_instance.backfill_acp_session_cwd()
if repaired:
logger.info("Backfilled cwd for %d ACP session(s) from model_config", repaired)
except Exception:
logger.debug("ACP session cwd backfill failed", exc_info=True)
return self._db_instance
def _persist(self, state: SessionState) -> None:
@@ -310,12 +331,21 @@ class SessionManager:
# Empty editor probes stay ephemeral; copied fork history persists.
return
db.create_session(session_id=state.session_id, source="acp", model=model_str,
model_config=session_meta)
model_config=session_meta, cwd=state.cwd or None)
else:
try:
db.update_session_meta(state.session_id, json.dumps(session_meta), model_str)
except Exception:
logger.debug("Failed to update ACP session metadata", exc_info=True)
# The create branch above is not the live path: an agent that owns
# persistence to this same DB flushes the transcript incrementally,
# so the row already exists by the time we get here.
# update_session_meta touches only model_config/model, so without
# this promotion the column stays NULL for the whole session and
# Desktop files it as unassigned. Claiming a generation keeps the
# A -> B -> A ordering contract shared with update_cwd().
if state.cwd:
self._schedule_git_metadata(state, self._claim_cwd_generation(state))
# An agent that owns persistence to this same DB already flushed the transcript
# incrementally (append_message) and keeps pre-compaction turns as archived
@@ -341,6 +371,50 @@ class SessionManager:
except Exception:
logger.warning("Failed to persist ACP session %s", state.session_id, exc_info=True)
def _claim_cwd_generation(self, state: SessionState) -> Optional[int]:
"""Write the cwd column and return its new git-metadata generation.
Returns None when the row does not exist yet (a contentless session is
deliberately ephemeral until it has history) or the DB is unavailable.
"""
db = self._get_db()
if db is None or not state.cwd:
return None
try:
return db.update_session_cwd(state.session_id, state.cwd)
except Exception:
logger.debug("Failed to persist ACP session cwd column for %s",
state.session_id, exc_info=True)
return None
def _schedule_git_metadata(self, state: SessionState, generation: Optional[int]) -> None:
"""Probe git off the critical path and publish under ``generation``.
``session/new`` is on the editor's interactive path; ``git rev-parse``
on a cold or networked filesystem is not something to put in front of
the user. The generation guard in ``publish_session_git_metadata``
means a slow probe for a previous workspace is dropped rather than
applied to the new one.
"""
if not generation or not state.cwd:
return
session_id, cwd = state.session_id, state.cwd
def _run() -> None:
try:
from tui_gateway import git_probe
branch, root = git_probe.branch(cwd), git_probe.common_repo_root(cwd)
if not (branch or root):
return
db = self._get_db()
if db is not None:
db.publish_session_git_metadata(session_id, cwd, generation, branch, root)
except Exception:
logger.debug("Failed to publish ACP git metadata for %s", session_id, exc_info=True)
threading.Thread(target=_run, name=f"acp-git-meta-{session_id[:8]}", daemon=True).start()
def _restore(self, session_id: str) -> Optional[SessionState]:
"""Load an ACP session from the database into memory, recreating the AIAgent."""
db = self._get_db()