fc589b6898
Follow-up to the #105749 + #123903 salvage: - A named profile without terminal.backend is local. Both contributor helpers fell back to the LAUNCH profile's backend, so a local profile opened from an ssh launch had its terminal.cwd treated as remote. - _bound_terminal_backend() is the single resolver (create, completion, workspace.move, display heal, settle-follow, terminal tool). The terminal tool now reads it too, so an ssh profile under a local launch keeps /home/kali instead of the display heal persisting /home. - _declared_remote_profile_cwd() only honours a profile that itself declares backend: ssh; the placeholder set is gone (the ~/absolute shape check already rejects ".", "auto", "cwd"). - One loader for a named profile's terminal section (_profile_terminal_cfg), shared with _profile_configured_cwd. - Eager row at session.create and "row cwd = explicit" on hydrate apply to remote sessions only; local project drafts stay lazy and keep settle-following, as on main. - config.get project forwards the pinned profile (and marks a picked path explicit) so the desktop gets the remote dir back; the renderer keeps adopting the server's normalized cwd for local users (WSL translation, abspath) instead of bypassing it. - Dropped #123903's cwd_explicit-decides-intent change in session.create: it made every local desktop new chat in a project lose its workspace and AGENTS.md.
629 lines
34 KiB
Python
629 lines
34 KiB
Python
"""Session working-directory + durable session row: cwd resolution/healing, session.db row ensure,
|
|
branch seed, history rewind, git meta persistence. Bodies are rebound onto server.py's globals at
|
|
install time (method_ctx.bind_module), so they reference server.py globals bare.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import contextlib
|
|
from tui_gateway import git_probe
|
|
|
|
from .method_ctx import bind_module
|
|
|
|
|
|
def _normalize_completion_path(path_part: str) -> str:
|
|
expanded = os.path.expanduser(path_part)
|
|
if os.name != "nt":
|
|
normalized = expanded.replace("\\", "/")
|
|
if len(normalized) >= 3 and normalized[1] == ":" and normalized[2] == "/" and normalized[0].isalpha():
|
|
return f"/mnt/{normalized[0].lower()}/{normalized[3:]}"
|
|
return expanded
|
|
|
|
|
|
def _completion_cwd(params: dict | None = None) -> str:
|
|
params = params or {}
|
|
# Provenance for the client-sent ``cwd`` (#52589): the desktop seeds a new chat's cwd
|
|
# from its app-global workspace (the launch profile's configured directory or the
|
|
# project scope) when the user did NOT pick one. That inherited default must NOT
|
|
# override a NAMED profile's own ``terminal.cwd`` — only a deliberate per-session
|
|
# workspace pick (``cwd_explicit``) wins over the profile config. Path equality
|
|
# cannot tell the two apart, so the desktop ships the flag alongside the path.
|
|
client_cwd = params.get("cwd")
|
|
if not params.get("cwd_explicit") and client_cwd:
|
|
profile_home = _profile_home(params.get("profile"))
|
|
profile_cwd = _profile_configured_cwd(profile_home)
|
|
if profile_cwd:
|
|
return profile_cwd
|
|
# An ssh profile's terminal.cwd is not on this host, so the isdir check above drops it.
|
|
if remote_cwd := _declared_remote_profile_cwd(profile_home):
|
|
return remote_cwd
|
|
# A session bound to another profile resolves its workspace from THAT profile's config before the launch profile's
|
|
# env var; the dashboard's in-memory gateway does NOT inherit the PTY child's bridged TERMINAL_CWD, so a configured
|
|
# terminal.cwd is read directly.
|
|
raw = (client_cwd or _sessions.get(params.get("session_id") or "", {}).get("cwd")
|
|
or _profile_configured_cwd(_profile_home(params.get("profile"))) or _launch_configured_cwd()
|
|
or os.environ.get("TERMINAL_CWD") or os.getcwd())
|
|
# The BOUND profile's backend, not the launch profile's: under multiplex HERMES_HOME is not yet rebound at
|
|
# session.create, so the process-global _effective_terminal_backend() would misread an ssh/docker profile as
|
|
# local and drop its remote cwd to getcwd().
|
|
# A non-local backend's cwd lives inside the target environment, not on the host: pass it raw (as
|
|
# _terminal_task_cwd_with_source does) instead of the local isdir gate that would drop it to getcwd().
|
|
if _bound_terminal_backend(_profile_home(params.get("profile"))) != "local":
|
|
return str(raw)
|
|
with contextlib.suppress(Exception):
|
|
resolved = os.path.abspath(os.path.expanduser(str(raw)))
|
|
if os.path.isdir(resolved):
|
|
return resolved
|
|
return os.getcwd()
|
|
|
|
|
|
def _workdir_terminal_cfg(key: str) -> str:
|
|
"""Stripped ``terminal.<key>`` from config, or "" when unset/unreadable."""
|
|
with contextlib.suppress(Exception):
|
|
terminal_cfg = _load_cfg().get("terminal", {})
|
|
if isinstance(terminal_cfg, dict):
|
|
return str(terminal_cfg.get(key) or "").strip()
|
|
return ""
|
|
|
|
|
|
def _bound_terminal_backend(profile_home) -> str:
|
|
"""Terminal backend of the profile a session/RPC is bound to: a named profile's own config, else the process's."""
|
|
return _profile_terminal_backend(Path(profile_home) if profile_home else None) or _effective_terminal_backend()
|
|
|
|
|
|
def _declared_remote_profile_cwd(profile_home) -> str | None:
|
|
"""A named ssh profile's own ``terminal.cwd`` (``~``, ``~/…`` or absolute), unchecked against this host.
|
|
|
|
``_profile_configured_cwd`` requires ``os.path.isdir``; an ssh working directory lives on the remote, so that
|
|
check drops it and the launch profile's ``TERMINAL_CWD`` wins. Only a profile that itself declares
|
|
``backend: ssh`` qualifies — placeholders (``.``/``auto``/``cwd``) fail the shape check.
|
|
"""
|
|
terminal = _profile_terminal_cfg(profile_home)
|
|
if str(terminal.get("backend") or "").strip().lower() != "ssh":
|
|
return None
|
|
raw = str(terminal.get("cwd") or "").strip()
|
|
return raw if raw == "~" or raw.startswith("~/") or os.path.isabs(raw) else None
|
|
|
|
|
|
def _terminal_task_cwd(session: dict | None) -> str:
|
|
"""The cwd terminal_tool should use for this TUI session (NOT host-validated: a non-local backend's cwd lives
|
|
inside the target environment)."""
|
|
return _terminal_task_cwd_with_source(session)[0]
|
|
|
|
|
|
def _terminal_task_cwd_with_source(session: dict | None) -> tuple[str, str]:
|
|
"""``(cwd, source)``: ``"session"`` for THIS session's workspace (``explicit_cwd``/tracked dir), ``"process"`` for
|
|
the global ``TERMINAL_CWD``/``terminal.cwd`` fallback — under per-session docker isolation that is a PREVIOUS
|
|
session's launch artifact, so terminal_tool refuses it as a bind-mount source."""
|
|
backend = _bound_terminal_backend((session or {}).get("profile_home"))
|
|
if backend != "local":
|
|
# THIS session's explicit workspace beats the LAST session's env var.
|
|
if session and session.get("explicit_cwd") and session.get("cwd"):
|
|
return str(session["cwd"]), "session"
|
|
# Process TERMINAL_CWD is the launch profile's; a named ssh profile's own terminal.cwd wins over it.
|
|
if remote_cwd := _declared_remote_profile_cwd((session or {}).get("profile_home")):
|
|
return remote_cwd, "session"
|
|
raw = os.environ.get("TERMINAL_CWD", "").strip() or _workdir_terminal_cfg("cwd")
|
|
if raw and raw not in {".", "auto", "cwd"}:
|
|
return raw, "process"
|
|
if backend == "ssh":
|
|
return "~", "process"
|
|
if session and session.get("cwd"):
|
|
return str(session["cwd"]), "session"
|
|
return _completion_cwd(), "process"
|
|
|
|
|
|
def _session_cwd(session: dict | None) -> str:
|
|
return str(session["cwd"]) if session and session.get("cwd") else _completion_cwd()
|
|
|
|
|
|
# Sources whose launch directory is an artifact of how the app was started, not a workspace the user picked.
|
|
_LAUNCH_CWD_NOT_A_WORKSPACE = {"desktop"}
|
|
|
|
|
|
def _context_cwd_is_launch_artifact(session: dict | None) -> bool:
|
|
"""Whether the session cwd came from app launch rather than user intent."""
|
|
return bool(session and not session.get("explicit_cwd") and _session_source(session) in _LAUNCH_CWD_NOT_A_WORKSPACE)
|
|
|
|
|
|
def _persisted_session_cwd(session: dict) -> str | None:
|
|
"""The cwd to stamp on the session's DB row, or None to leave it unset (launch-dir rule: ``_ensure_session_db_row``)."""
|
|
if session.get("explicit_cwd"):
|
|
return _session_cwd(session)
|
|
if _session_source(session) in _LAUNCH_CWD_NOT_A_WORKSPACE:
|
|
return None
|
|
return str(session.get("cwd") or "") or None # the session's OWN dir, never _session_cwd's gateway-wide fallback
|
|
|
|
|
|
def _heal_dead_cwd(cwd: str) -> str:
|
|
"""Resolve a session cwd inside a now-deleted directory (e.g. a removed linked worktree, which probes to no branch
|
|
while the sidebar folds it to the main lane): walk up to the first existing ancestor and take its common git root.
|
|
Local backends only — a remote/SSH cwd may legitimately not exist on the host, so callers skip healing there."""
|
|
raw = (cwd or "").strip()
|
|
if not raw or os.path.isdir(raw):
|
|
return raw
|
|
probe = raw
|
|
for _ in range(64):
|
|
parent = os.path.dirname(probe)
|
|
if not parent or parent == probe:
|
|
break
|
|
probe = parent
|
|
if os.path.isdir(probe):
|
|
break
|
|
if not os.path.isdir(probe):
|
|
return raw
|
|
with contextlib.suppress(Exception):
|
|
return git_probe.common_repo_root(probe) or git_probe.repo_root(probe) or probe
|
|
return probe
|
|
|
|
|
|
def _is_local_terminal_backend() -> bool:
|
|
backend = (os.environ.get("TERMINAL_ENV") or "").strip().lower()
|
|
return not backend or backend == "local"
|
|
|
|
|
|
def _session_is_local_backend(session: dict | None) -> bool:
|
|
"""Whether THIS session's terminal backend is local. A multiplexed gateway serves many profiles from
|
|
ONE process, so the env-only _is_local_terminal_backend() reports the LAUNCH profile (usually local)
|
|
for a session actually bound to an ssh/docker profile - which then heals its remote cwd to a host
|
|
ancestor (/home) and persists that. Read the BOUND profile's backend first (profile_home), falling
|
|
back to the process env only when the session carries no profile."""
|
|
# The process fallback reads env OR config: a per-profile gateway (hermes -p felix) sets terminal.backend=ssh
|
|
# in config but leaves TERMINAL_ENV unset, so an env-only check would heal a live remote cwd to /home.
|
|
return _bound_terminal_backend(session.get("profile_home") if session else None) == "local"
|
|
|
|
|
|
def _effective_terminal_backend() -> str:
|
|
"""Active terminal backend name (``local``, ``docker``, ``ssh``, ...): ``TERMINAL_ENV`` when set (launchers bridge
|
|
``terminal.backend`` into env), else the ``terminal.backend`` config key (in-process gateways skip that bridge)."""
|
|
backend = (os.environ.get("TERMINAL_ENV") or "").strip().lower()
|
|
if not backend or backend == "local":
|
|
backend = _workdir_terminal_cfg("backend").lower()
|
|
return backend or "local"
|
|
|
|
|
|
def _display_session_cwd(session: dict | None) -> str:
|
|
"""Session cwd for display/probe surfaces, healed past deleted worktrees (healed value persisted back; local only)."""
|
|
cwd = _session_cwd(session)
|
|
if not _session_is_local_backend(session):
|
|
return cwd
|
|
healed = _heal_dead_cwd(cwd)
|
|
if healed and healed != cwd and session is not None:
|
|
session["cwd"] = healed
|
|
_persist_session_cwd_and_schedule_git_meta(session, healed)
|
|
return healed
|
|
|
|
|
|
def _reconcile_session_cwd_from_terminal(session: dict | None) -> bool:
|
|
"""Re-anchor a session that SETTLED in another worktree of the SAME repo. Returns moved. An agent told to work in
|
|
a fresh worktree `git worktree add`s and `cd`s in while the session stays pinned (labelled with the primary
|
|
checkout's branch). A plain `cd` is deliberately NOT a workspace move (see ``_apply_project_workspace``): a non-git
|
|
workspace stepping into a repo or a visit to an unrelated repo is browsing, and an explicitly chosen workspace is
|
|
never overridden. Local backends only (a remote cwd cannot be stat'ed or git-probed here)."""
|
|
# An explicit choice only moves by another explicit action; a cwd adopted HERE is marked `cwd_from_settle` so
|
|
# successive settles keep following.
|
|
if not session or not _session_is_local_backend(session):
|
|
return False
|
|
if session.get("explicit_cwd") and not session.get("cwd_from_settle"):
|
|
return False
|
|
try:
|
|
from tools.terminal_tool import get_session_cwd
|
|
if not (recorded := get_session_cwd(session.get("session_key") or "")):
|
|
return False
|
|
except Exception:
|
|
return False
|
|
resolved = os.path.abspath(os.path.expanduser(str(recorded)))
|
|
current = os.path.abspath(os.path.expanduser(_session_cwd(session)))
|
|
if resolved == current or not os.path.isdir(resolved):
|
|
return False
|
|
# Worktree ROOTS (folding to the common root would hide the move), both in a git tree, different from each other,
|
|
# sharing the SAME common .git dir.
|
|
landed, current_root = git_probe.repo_root(resolved), git_probe.repo_root(current)
|
|
if not landed or not current_root or landed == current_root:
|
|
return False
|
|
landed_common = git_probe.common_repo_root(resolved)
|
|
if not landed_common or landed_common != git_probe.common_repo_root(current):
|
|
return False
|
|
# This is the session's workspace now (a desktop launch-artifact cwd earns a real row); the settle marker keeps it
|
|
# overridable by the NEXT settle.
|
|
session.update(cwd=resolved, explicit_cwd=True, cwd_from_settle=True)
|
|
_register_session_cwd(session)
|
|
_persist_session_cwd_and_schedule_git_meta(session, resolved)
|
|
return True
|
|
|
|
|
|
def _emit_settled_session_info(sid: str, session: dict, agent) -> None:
|
|
"""Emit end-of-turn ``session.info``, reconciling a settled cwd first (the agent has stopped moving; riding the
|
|
turn-end event needs no new event type/round trip)."""
|
|
try:
|
|
_reconcile_session_cwd_from_terminal(session)
|
|
except Exception:
|
|
logger.debug("failed to reconcile settled session cwd", exc_info=True)
|
|
_emit("session.info", sid, _session_info(agent, session))
|
|
|
|
|
|
def _session_source(session: dict | None) -> str:
|
|
source = str(session.get("source") or "").strip() if session else ""
|
|
return source or _resolve_session_platform()
|
|
|
|
|
|
def _register_session_cwd(session: dict | None) -> None:
|
|
if not session:
|
|
return
|
|
# Workspace moves must reach lazy/restarted runtimes, not just terminal tools.
|
|
# Do not reinitialize memory providers or invalidate the cached system prompt.
|
|
if hasattr(agent := session.get("agent"), "session_cwd"):
|
|
agent.session_cwd = session.get("cwd") or None
|
|
with contextlib.suppress(Exception):
|
|
from tools.terminal_tool import register_task_env_overrides
|
|
cwd, cwd_source = _terminal_task_cwd_with_source(session)
|
|
register_task_env_overrides(session["session_key"], {"cwd": cwd, "cwd_source": cwd_source})
|
|
|
|
|
|
def _workdir_row_model_config(session: dict) -> tuple[str, dict]:
|
|
"""``(model, model_config)`` for a fresh session row. The session's own model/effort/fast pick (composer override
|
|
or restored /model switch) must own the row: the agent isn't built yet at first prompt.submit, and writing the
|
|
global default here wins the INSERT-OR-IGNORE race (a reconnect silently reverts to the profile default).
|
|
model_config carries provider/reasoning/service_tier so resume restores effort + fast too."""
|
|
override = raw if isinstance(raw := session.get("model_override"), dict) else {}
|
|
row_model = str(override.get("model") or "").strip() or _session_default_model(session)
|
|
model_config: dict = {k: str(v) for k in ("model", "provider", "base_url", "api_mode") if (v := override.get(k))}
|
|
# A RESOLVED provider "custom" (named ``providers:``/``custom_providers:`` entry) persisted bare here is the origin
|
|
# of "No LLM provider configured" rows (resume routes to OpenRouter with no key). Recover the durable
|
|
# ``custom:<name>`` identity (matches _runtime_model_config).
|
|
if str(model_config.get("provider") or "").strip().lower() == "custom":
|
|
try:
|
|
from hermes_cli.runtime_provider import canonical_custom_identity
|
|
healed = canonical_custom_identity(
|
|
base_url=model_config.get("base_url") or None, model=model_config.get("model") or row_model or None)
|
|
if healed:
|
|
model_config["provider"] = healed
|
|
except Exception:
|
|
logger.debug("custom provider identity recovery failed (db row)", exc_info=True)
|
|
if (reasoning := session.get("create_reasoning_override")) is not None:
|
|
model_config["reasoning_config"] = reasoning
|
|
if (service_tier := session.get("create_service_tier_override")) is not None:
|
|
# "" is the in-memory sentinel for an explicit normal tier (bypasses _make_agent's profile fallback); persist a
|
|
# durable marker so resume can tell it from an inherited tier.
|
|
model_config["service_tier"] = service_tier or "normal"
|
|
# Same ``_branched_from`` marker the TUI /branch uses (list_sessions_rich + sidebar nesting).
|
|
if parent_session_id := session.get("parent_session_id"):
|
|
model_config["_branched_from"] = parent_session_id
|
|
# Room plumbing always follows the member profile. Canonical Bot Chats do too until the composer records an
|
|
# explicit chat-scoped pick plus the profile model it diverged from (see _stored_session_runtime_overrides).
|
|
for flag in ("room_plumbing", "follow_profile_config"):
|
|
if session.get(flag):
|
|
model_config[flag] = True
|
|
if isinstance(composer_profile := session.get("composer_override_profile"), dict):
|
|
model_config["composer_override_profile"] = composer_profile
|
|
return row_model, model_config
|
|
|
|
|
|
def _ensure_session_db_row(session: dict) -> bool:
|
|
"""Idempotently persist the session's DB row on first real activity (prompt.submit), so abandoned drafts never
|
|
leave an empty "Untitled" session. INSERT OR IGNORE: re-calls and the AIAgent's lazy create are no-ops. Returns
|
|
False only when the store is unavailable (no openable state.db) — prompt.submit fails the send loudly instead of
|
|
streaming into a store that will never save it; no key / best-effort / success are all True.
|
|
|
|
A cwd the user *chose* is always persisted. Otherwise the launch directory stands in only for terminal sessions
|
|
(the user deliberately ``cd``'d there; dropping it left the sidebar with no cwd AND no git_repo_root); desktop
|
|
launch dirs (``/``, home) stay null -> "No workspace".
|
|
|
|
See #98924.
|
|
"""
|
|
if not (key := session.get("session_key")):
|
|
return
|
|
# Persist into the session's own profile db (global remote mode), not the launch profile's — otherwise the unified
|
|
# list mis-tags the row and resume 404s ("session not found").
|
|
profile_home = session.get("profile_home")
|
|
with _workdir_owner_db(session, "failed to open profile db for session row") as db:
|
|
if db is _WORKDIR_DB_OPEN_FAILED:
|
|
return False
|
|
if db is None:
|
|
# Fail loud ONLY when the store failed to open (_db_error records the SessionDB open exception); None with
|
|
# no recorded error means "no store in this context" -> True.
|
|
# A None db with no recorded error means "no store in this context" (degraded harness, store
|
|
# deliberately absent) — that keeps the pinned best-effort contract and stays True. See #98924.
|
|
return _db_error is None
|
|
row_model, model_config = _workdir_row_model_config(session)
|
|
try:
|
|
persisted_cwd = _persisted_session_cwd(session)
|
|
db.create_session(
|
|
key, source=_session_source(session), model=row_model, model_config=model_config or None,
|
|
parent_session_id=session.get("parent_session_id") or None, cwd=persisted_cwd,
|
|
# The login this session was opened under, in the same ``<provider>:<id>`` form the agent is
|
|
# built with — the row is the only place the identity reaches the store, and the upsert can't
|
|
# add it later (user_id is set at insert). None (no password provider, legacy token, stdio)
|
|
# leaves the column empty exactly as before.
|
|
user_id=_session_auth_user_id(session),
|
|
# Self-describing rows: aggregators merging several profile DBs can't rely on which file a row came
|
|
# from; a NULL is only repaired by the one-shot backfill.
|
|
# Stamp the launch profile explicitly instead of leaving NULL — NULL is exactly what the
|
|
# #94724 legacy-owner backfill exists to repair, and rows minted AFTER that one-shot
|
|
# backfill ran stayed NULL forever: profile-keyed matching then drops them from the sidebar
|
|
# and deep links can't resolve them (#99222).
|
|
profile_name=profile_name_for_home(profile_home) or _current_profile_name())
|
|
# create_session is INSERT OR IGNORE: if the AIAgent's lazy create already minted this row WITHOUT a
|
|
# cwd (a non-local project session, where the cwd is only known here), the create above is a no-op and
|
|
# the cwd never lands -> the sidebar tree drops the session to Home. Force the chosen cwd onto the row.
|
|
if persisted_cwd:
|
|
with contextlib.suppress(Exception):
|
|
db.update_session_cwd(key, persisted_cwd)
|
|
# Born hidden (session.create hidden=true, or set_hidden before the row existed): apply the deferred intent.
|
|
if session.get("pending_hidden"):
|
|
try:
|
|
if db.set_session_hidden(key, True):
|
|
session.pop("pending_hidden", None)
|
|
except Exception:
|
|
logger.debug("failed to apply pending hidden flag", exc_info=True)
|
|
except Exception as exc:
|
|
# Disk-full is not a soft failure: swallowed here, prompt.submit returns {"status":"streaming"} and the
|
|
# message vanishes silently.
|
|
_workdir_reraise_disk_full(exc, "failed to persist desktop session row")
|
|
return True
|
|
|
|
|
|
def _workdir_reraise_disk_full(exc: BaseException, log_msg: str) -> None:
|
|
"""Re-raise a disk-full write error (the caller must surface it); debug-log the rest."""
|
|
from hermes_state_errors import is_disk_full_error
|
|
if is_disk_full_error(exc):
|
|
raise exc
|
|
logger.debug(log_msg, exc_info=True)
|
|
|
|
|
|
# Seed row fields copied from the parent transcript. display_kind/metadata: timeline markers ride as role=user;
|
|
# dropping the tag re-plants them as bare user turns after a restart and corrupts the truncate ordinal address space.
|
|
_WORKDIR_SEED_FIELDS = (
|
|
"content", "reasoning", "reasoning_content", "reasoning_details", "codex_reasoning_items",
|
|
"codex_message_items", "display_kind", "display_metadata", "timestamp")
|
|
|
|
|
|
def _persist_branch_seed(session: dict) -> None:
|
|
"""Persist a seeded transcript once its row exists. Seeded messages (a branch's copied parent, a client's
|
|
opening turns) live only in ``session["history"]`` (ridden into the agent as ``conversation_history``, which
|
|
``_flush_messages_to_session_db`` skips by identity), so the row would otherwise resume without them. Runs
|
|
once: at create for a seeded session, else at the first submit after ``_ensure_session_db_row`` wrote the
|
|
row. ``seeded`` is stamped by session.create; a resumed session carries its stored transcript in
|
|
``history`` and must never re-append it."""
|
|
if not (key := session.get("session_key")) or not session.get("seeded") or session.get("_branch_seed_persisted"):
|
|
return
|
|
with session["history_lock"]:
|
|
seed = [dict(msg) for msg in (session.get("history") or [])]
|
|
if not seed:
|
|
return
|
|
with _session_db(session) as db:
|
|
if db is None:
|
|
return
|
|
try:
|
|
# Chunked so each BEGIN IMMEDIATE stays short (a seed can be hundreds of rows); a mid-copy failure leaves a
|
|
# partial seed with _branch_seed_persisted unset.
|
|
# Bounded-chunk transactions (see #23254): a branch seed can be hundreds of rows; chunking keeps
|
|
# each BEGIN IMMEDIATE short so concurrent writers aren't starved.
|
|
db.append_messages_batch(
|
|
key, [{"role": msg.get("role", "user"), **{f: msg.get(f) for f in _WORKDIR_SEED_FIELDS}} for msg in seed],
|
|
chunk_rows=500)
|
|
session["_branch_seed_persisted"] = True
|
|
except Exception as exc:
|
|
_workdir_reraise_disk_full(exc, "branch seed persist failed")
|
|
|
|
|
|
def _write_submit_user_row(session: dict, text: Any, display_kind: str | None) -> dict | None:
|
|
"""Write the submitted user turn to the transcript and RETURN the durable dict (stamped
|
|
``_DB_PERSISTED_MARKER``/``_row_id``) WITHOUT slotting it on the session. The write half of
|
|
:func:`_persist_submit_user_row`, shared by the busy-queue accept (which attaches the dict to
|
|
the queue envelope, never the shared session slot a possibly-still-staged in-flight turn owns).
|
|
Returns None when nothing was written (no key / non-text / store unavailable / failed write)."""
|
|
key = session.get("session_key")
|
|
if not key or not isinstance(text, str) or not text.strip():
|
|
return None
|
|
from agent.context_compressor import _DB_PERSISTED_MARKER
|
|
from agent.message_metadata import stamp_message_timestamp
|
|
staged = stamp_message_timestamp({"role": "user", "content": text})
|
|
if display_kind:
|
|
staged["display_kind"] = display_kind
|
|
with _session_db(session) as db:
|
|
if db is None:
|
|
return None
|
|
try:
|
|
staged["_row_id"] = db.append_message(
|
|
key, "user", content=text, display_kind=display_kind, timestamp=staged["timestamp"])
|
|
except Exception as exc:
|
|
_workdir_reraise_disk_full(exc, "submit-time user row persist failed")
|
|
return None
|
|
staged[_DB_PERSISTED_MARKER] = True
|
|
return staged
|
|
|
|
|
|
def _persist_submit_user_row(session: dict, text: Any, display_kind: str | None) -> None:
|
|
"""Write the submitted user turn at send time, before the agent build and turn: the agent's own
|
|
crash persist only runs once the build finished, so quitting a frozen app during a slow first build
|
|
left a session row with no message (#111868). The dict is staged on the session already stamped
|
|
durable (the shape ``quiet_single_query`` re-stages an unanswered DM in) so the turn adopts it via
|
|
``_stage_turn_user_message`` and the flush writes no second row. A failed write stages nothing:
|
|
the turn's crash persist then writes the row as before."""
|
|
session.pop("_submit_user_row", None) # a failed/unsupported write must not acknowledge an older send
|
|
if (staged := _write_submit_user_row(session, text, display_kind)) is not None:
|
|
session["_submit_user_row"] = staged
|
|
|
|
|
|
def _adopt_submit_user_row(session: dict, agent, persist_user_message: Any, text: Any) -> None:
|
|
"""Hand the row written at submit to the turn as its user dict (``agent._pending_cli_user_message``,
|
|
adopted by ``_stage_turn_user_message`` when the content matches). A prompt the prologue rewrote
|
|
(@-expansion, image parts) first updates that row so the durable transcript replays what the model
|
|
was sent and the ``api_content`` sidecar can address it; ``_row_id`` rides along for that stamp.
|
|
``text`` is THIS turn's raw submit: a staged row from an earlier send (its turn ended before the agent
|
|
ran) is discarded untouched, so the DB row stays the user's message and never a synthesized turn's text."""
|
|
staged = session.pop("_submit_user_row", None)
|
|
if not isinstance(staged, dict) or agent is None or staged.get("content") != text:
|
|
return
|
|
if staged["content"] != persist_user_message:
|
|
from agent.session_persistence import _durable_content
|
|
with _session_db(session) as db:
|
|
if db is None:
|
|
return
|
|
try:
|
|
db.set_user_message_content(
|
|
session["session_key"], staged["_row_id"], _durable_content(persist_user_message))
|
|
except Exception:
|
|
logger.debug("submit-time user row update failed; the turn writes its own row", exc_info=True)
|
|
return
|
|
staged["content"] = persist_user_message
|
|
from agent.session_persistence import _persist_lock
|
|
with _persist_lock(agent):
|
|
agent._pending_cli_user_message = staged
|
|
|
|
|
|
# Yielded by _workdir_owner_db when the profile db failed to OPEN (vs "no store in this context"); row creation fails loud.
|
|
_WORKDIR_DB_OPEN_FAILED = object()
|
|
|
|
|
|
@contextlib.contextmanager
|
|
def _workdir_owner_db(session: dict, fail_log: str):
|
|
"""Body of :func:`_session_db`; ``_ensure_session_db_row`` uses it directly so a patched ``_session_db`` can't alter rows."""
|
|
db, close_db = None, False
|
|
if profile_home := session.get("profile_home"):
|
|
try:
|
|
from hermes_state_registry import acquire
|
|
db, close_db = acquire(Path(profile_home) / "state.db"), True
|
|
except Exception:
|
|
logger.debug(fail_log, exc_info=True)
|
|
db = _WORKDIR_DB_OPEN_FAILED
|
|
else:
|
|
db = _get_db()
|
|
try:
|
|
yield db
|
|
finally:
|
|
if close_db and db is not None:
|
|
with contextlib.suppress(Exception):
|
|
from hermes_state_registry import release_or_close
|
|
release_or_close(db)
|
|
|
|
|
|
@contextlib.contextmanager
|
|
def _session_db(session: dict):
|
|
"""Yield the SessionDB that owns this session's row (profile-aware): a remote/profile session persists into its own
|
|
profile's ``state.db`` (fresh handle, closed on exit); else the shared ``_get_db()`` handle (left open). None if unavailable."""
|
|
with _workdir_owner_db(session, "failed to open profile db for session") as db:
|
|
yield None if db is _WORKDIR_DB_OPEN_FAILED else db
|
|
|
|
|
|
def _rewind_active_session_history(
|
|
session: dict, user_ordinal: int, *, require_retryable: bool = False) -> tuple[list[dict], dict, int]:
|
|
"""Rewind one canonical user turn while retaining carrier scaffolding. Caller holds ``history_lock``. Persistent
|
|
sessions go through ``SessionDB.rewind_user_turn`` (the durable transcript is the authority; memory is installed
|
|
only after the commit); a session without a key rewinds the warm history alone."""
|
|
from agent.context_compressor import history_before_user_originated_turn, retryable_user_text, user_originated_turn_view
|
|
|
|
history = _history_without_ephemeral_scaffolding(session.get("history", []))
|
|
user_indices = [i for i, m in enumerate(history) if user_originated_turn_view(m) is not None]
|
|
if user_ordinal < 0 or user_ordinal >= len(user_indices):
|
|
raise ValueError("target user message is no longer in session history")
|
|
session_key = str(session.get("session_key") or "").strip()
|
|
if session_key:
|
|
with _session_db(session) as db:
|
|
if db is None:
|
|
raise RuntimeError("session database is unavailable")
|
|
outcome = db.rewind_user_turn(
|
|
session_key, user_ordinal, warm_history=history, require_retryable=require_retryable,
|
|
adopt_row_ids=True)
|
|
installed, live_view, rewound_count = outcome.prefix, outcome.live_view, outcome.rewound_count
|
|
else:
|
|
target_index = user_indices[user_ordinal]
|
|
installed, live_view = history_before_user_originated_turn(history, target_index)
|
|
rewound_count = len(history) - target_index
|
|
if require_retryable:
|
|
retryable_user_text(live_view.get("content"))
|
|
|
|
installed = [message.copy() for message in installed]
|
|
session["history"] = installed
|
|
session["history_version"] = int(session.get("history_version", 0)) + 1
|
|
agent = session.get("agent")
|
|
if agent is not None:
|
|
agent._session_messages = installed
|
|
if hasattr(agent, "_last_flushed_db_idx"):
|
|
agent._last_flushed_db_idx = len(installed) if session_key else 0
|
|
if hasattr(agent, "_db_flush_scan_prefix"):
|
|
agent._db_flush_scan_prefix = installed[:] if session_key else None
|
|
return installed, live_view, rewound_count
|
|
|
|
|
|
def _history_without_ephemeral_scaffolding(history: list[dict]) -> list[dict]:
|
|
"""Return the durable transcript shape without transient recovery rows."""
|
|
from agent.session_persistence import _is_ephemeral_scaffolding
|
|
return [message.copy() for message in history if not _is_ephemeral_scaffolding(message)]
|
|
|
|
|
|
def _workdir_valid_generation(generation) -> bool:
|
|
"""A claimed DB probe generation: a positive int (bool excluded)."""
|
|
return not isinstance(generation, bool) and isinstance(generation, int) and generation >= 1
|
|
|
|
|
|
def _persist_session_git_meta(session: dict, cwd: str, generation: int) -> None:
|
|
"""Resolve + persist a session's git branch / repo root on a daemon thread: inline ``git`` probes on the
|
|
session-init / cwd-set path would stall startup on a slow or unreachable ``cwd``. Persists via the same
|
|
profile-aware db the caller wrote ``cwd`` to. Best-effort: a probe failure leaves the enrichment columns unset."""
|
|
session_key = session.get("session_key", "")
|
|
if not session_key or not cwd or not _workdir_valid_generation(generation):
|
|
return
|
|
# Snapshot routing fields; the live session dict may be gone when the thread runs.
|
|
db_session = {"session_key": session_key, "profile_home": session.get("profile_home")}
|
|
|
|
def _run() -> None:
|
|
try:
|
|
branch, root = git_probe.branch(cwd), git_probe.common_repo_root(cwd)
|
|
if not (branch or root):
|
|
return
|
|
with _session_db(db_session) as db:
|
|
if db is not None:
|
|
db.publish_session_git_metadata(session_key, cwd, generation, branch, root)
|
|
except Exception:
|
|
logger.debug("failed to persist session git metadata", exc_info=True)
|
|
|
|
threading.Thread(target=_run, name="git-meta", daemon=True).start()
|
|
|
|
|
|
def _persist_session_cwd_and_schedule_git_meta(session: dict, cwd: str, *, db=None) -> int | None:
|
|
"""Claim a DB-backed probe generation, then start Git enrichment."""
|
|
try:
|
|
with (contextlib.nullcontext(db) if db is not None else _session_db(session)) as owner_db:
|
|
if owner_db is None:
|
|
return None
|
|
generation = owner_db.update_session_cwd(session.get("session_key", ""), cwd)
|
|
except Exception:
|
|
logger.debug("failed to persist session cwd", exc_info=True)
|
|
return None
|
|
if not _workdir_valid_generation(generation):
|
|
return None
|
|
_persist_session_git_meta(session, cwd, generation)
|
|
return generation
|
|
|
|
|
|
def _set_session_cwd(session: dict, cwd: str) -> str:
|
|
from hermes_constants import translate_cwd_for_wsl_backend
|
|
cwd = translate_cwd_for_wsl_backend(str(cwd))
|
|
resolved = os.path.abspath(os.path.expanduser(cwd))
|
|
# A non-local backend's workspace lives inside the target environment, not on the host: the local
|
|
# isdir gate would reject a valid remote project dir ("working directory does not exist"). Trust the
|
|
# bound profile's backend and store the path raw, mirroring _completion_cwd's non-local exemption.
|
|
if not _session_is_local_backend(session):
|
|
session.update(cwd=cwd, explicit_cwd=True, cwd_from_settle=False)
|
|
_register_session_cwd(session)
|
|
_persist_session_cwd_and_schedule_git_meta(session, cwd)
|
|
return cwd
|
|
if not os.path.isdir(resolved):
|
|
raise ValueError(f"working directory does not exist: {cwd}")
|
|
# An explicit user choice: persisted as the workspace (not the launch-dir fallback), superseding a settle-adopted cwd.
|
|
session.update(cwd=resolved, explicit_cwd=True, cwd_from_settle=False)
|
|
_register_session_cwd(session)
|
|
# The synchronous DB write claims ordering authority; git probes may publish only for that exact generation.
|
|
_persist_session_cwd_and_schedule_git_meta(session, resolved)
|
|
with contextlib.suppress(Exception):
|
|
from tools.terminal_tool_lifecycle import cleanup_vm
|
|
cleanup_vm(session["session_key"])
|
|
return resolved
|
|
|
|
|
|
def register(server) -> None:
|
|
"""Publish this module's helpers onto ``server``, rebound to its globals."""
|
|
bind_module(globals(), server, skip=("_",))
|