wehub-resource-sync
1353 lines
55 KiB
Python
1353 lines
55 KiB
Python
"""Runner-side session resource registry.
|
|
|
|
Authoritative owner/facade for all session-scoped resources: the
|
|
primary OS environment, terminal instances, and terminal-specific
|
|
environments. The public server API and runner-local tools call
|
|
this registry rather than reaching into ``TerminalRegistry`` or
|
|
``create_os_environment()`` directly.
|
|
|
|
See ``designs/SESSION_RESOURCES_API_DESIGN.md`` §Runner internal model.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import logging
|
|
import os
|
|
import threading
|
|
import time
|
|
from collections.abc import Callable
|
|
from dataclasses import dataclass
|
|
from enum import Enum
|
|
from pathlib import Path
|
|
from typing import TYPE_CHECKING, Any, Literal
|
|
|
|
from cachetools import TTLCache
|
|
|
|
from omnigent.entities.pagination import PagedList
|
|
from omnigent.entities.session_resources import (
|
|
DEFAULT_ENVIRONMENT_ID,
|
|
SessionResourceView,
|
|
filter_resources_by_type,
|
|
get_resource_by_id,
|
|
list_session_resources_from_terminal_registry,
|
|
terminal_environment_resource_id,
|
|
terminal_resource_id,
|
|
terminal_resource_view,
|
|
)
|
|
|
|
if TYPE_CHECKING:
|
|
from omnigent.inner.os_env import OSEnvironment
|
|
from omnigent.terminals.registry import TerminalRegistry
|
|
|
|
_logger = logging.getLogger(__name__)
|
|
|
|
_DEFAULT_WORKSPACE_ROOT = os.path.join(
|
|
os.environ.get("TMPDIR", "/tmp"),
|
|
"omnigent-sessions",
|
|
)
|
|
|
|
CODEX_NATIVE_TERMINAL_ROLE = "codex-native"
|
|
CLAUDE_NATIVE_TERMINAL_ROLE = "claude-native"
|
|
PI_NATIVE_TERMINAL_ROLE = "pi-native"
|
|
OPENCODE_NATIVE_TERMINAL_ROLE = "opencode-native"
|
|
CURSOR_NATIVE_TERMINAL_ROLE = "cursor-native"
|
|
KIRO_NATIVE_TERMINAL_ROLE = "kiro-native"
|
|
GOOSE_NATIVE_TERMINAL_ROLE = "goose-native"
|
|
# Role marker for the runner-owned native Antigravity (agy) TUI terminal.
|
|
# A generic terminal launched with ``terminal=antigravity`` shares the same
|
|
# public resource id, so the ensure path uses this private marker to tell a
|
|
# runner-owned agy TUI apart from an arbitrary terminal before reusing it.
|
|
ANTIGRAVITY_NATIVE_TERMINAL_ROLE = "antigravity-native"
|
|
QWEN_NATIVE_TERMINAL_ROLE = "qwen-native"
|
|
KIMI_NATIVE_TERMINAL_ROLE = "kimi-native"
|
|
HERMES_NATIVE_TERMINAL_ROLE = "hermes-native"
|
|
# Role marker for the embedded Omnigent REPL terminal auto-created for
|
|
# runner-hosted SDK sessions (``omnigent attach`` in a tmux pane — the
|
|
# SDK mirror of the native terminals above). The attach WebSocket uses
|
|
# this marker to recreate the terminal when its tmux session has died
|
|
# (the REPL exited or crashed) instead of rejecting the attach.
|
|
OMNIGENT_REPL_TERMINAL_ROLE = "omnigent-repl"
|
|
|
|
_IS_ALIVE_CACHE_TTL_S = 2.0
|
|
_IS_ALIVE_CACHE_MAX = 256
|
|
|
|
# Diff-track idle threshold (seconds) for the claude-native agent
|
|
# terminal's status watcher. Claude Code redraws its busy line every
|
|
# ~200ms while a turn is in progress, so a poll that sees no pane change
|
|
# only happens once Claude has actually stopped — making a short
|
|
# threshold safe against mid-turn false-idle. Kept distinct from the
|
|
# generic terminal-activity watcher's longer default so the session's
|
|
# "Working…" indicator flips to idle promptly (~1s) after Claude stops,
|
|
# matching the responsiveness of the hook-based ``Stop`` edge it
|
|
# replaces.
|
|
_CLAUDE_NATIVE_STATUS_IDLE_THRESHOLD_SECONDS = 1.0
|
|
|
|
# Poll interval (seconds) for the claude-native agent terminal's status
|
|
# watcher. Tighter than the generic terminal-activity watcher's 1s so the
|
|
# session's running/idle transitions feel responsive (~200ms) and so a
|
|
# pane change can be attributed to a recent client interaction within a
|
|
# tight window. Applied ONLY to this watcher (gated by the role) so we
|
|
# don't 5x the capture-pane subprocess load on every terminal.
|
|
_CLAUDE_NATIVE_STATUS_POLL_INTERVAL_SECONDS = 0.2
|
|
|
|
# Minimum wall-clock interval (seconds) between consecutive
|
|
# ``session.terminal.activity`` emissions for a single terminal. The
|
|
# claude-native agent terminal polls its pane every 200ms
|
|
# (:data:`_CLAUDE_NATIVE_STATUS_POLL_INTERVAL_SECONDS`) and Claude redraws
|
|
# its busy line on nearly every poll, so emitting an event on every
|
|
# pane-changed tick would push ~5 activity events/second for the whole
|
|
# turn. The web only needs a pulse inside its 1.5s "active" window
|
|
# (``ACTIVE_OUTPUT_WINDOW_MS`` in ``useTerminalStatuses``) to keep the
|
|
# badge lit, so coalescing to at most one emit per second cuts the event
|
|
# volume ~5x while keeping the badge solid. Generic terminals already poll
|
|
# at 1s, so this throttle is a no-op for them (their own poll spacing
|
|
# already exceeds the threshold).
|
|
_TERMINAL_ACTIVITY_EMIT_MIN_INTERVAL_SECONDS = 1.0
|
|
_TERMINAL_EXIT_OUTPUT_MAX_LINES = 40
|
|
_TERMINAL_EXIT_OUTPUT_MAX_CHARS = 4000
|
|
|
|
|
|
class TerminalLifecycle(Enum):
|
|
"""Session-lifecycle relationship for a terminal resource."""
|
|
|
|
REQUIRED = "required"
|
|
AUXILIARY = "auxiliary"
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class TerminalExitEvent:
|
|
"""Terminal exit event emitted by :class:`SessionResourceRegistry`.
|
|
|
|
:param session_id: Owning session/conversation identifier.
|
|
:param terminal_id: Opaque terminal resource id.
|
|
:param terminal_name: Terminal spec/resource name.
|
|
:param session_key: Per-launch terminal key.
|
|
:param lifecycle: Required/auxiliary lifecycle relationship.
|
|
:param command: Executable launched in the terminal, if known.
|
|
:param args_count: Number of arguments passed to the executable, if known.
|
|
The event intentionally does not expose argv contents because terminal
|
|
specs may contain credentials or other launch-only secrets.
|
|
:param cwd: Working directory used to launch the terminal, if known.
|
|
:param last_output: Last visible pane text captured before exit, if any.
|
|
:param session_was_idle: Whether the session's last PTY-derived status was
|
|
``idle`` at exit. ``True`` marks a clean shutdown after the turn
|
|
finished; ``False`` (the default — last seen ``running``, or never
|
|
observed) keeps a mid-turn crash or boot failure a failure.
|
|
"""
|
|
|
|
session_id: str
|
|
terminal_id: str
|
|
terminal_name: str
|
|
session_key: str
|
|
lifecycle: TerminalLifecycle
|
|
command: str | None = None
|
|
args_count: int | None = None
|
|
cwd: str | None = None
|
|
last_output: str | None = None
|
|
session_was_idle: bool = False
|
|
|
|
|
|
def _trim_terminal_exit_output(text: str | None) -> str | None:
|
|
"""Bound terminal-output diagnostics so a failure report stays compact."""
|
|
if text is None:
|
|
return None
|
|
stripped = text.strip()
|
|
if not stripped:
|
|
return None
|
|
lines = stripped.splitlines()
|
|
if len(lines) > _TERMINAL_EXIT_OUTPUT_MAX_LINES:
|
|
lines = [
|
|
f"... omitted {len(lines) - _TERMINAL_EXIT_OUTPUT_MAX_LINES} earlier line(s) ...",
|
|
*lines[-_TERMINAL_EXIT_OUTPUT_MAX_LINES:],
|
|
]
|
|
clipped = "\n".join(lines)
|
|
if len(clipped) > _TERMINAL_EXIT_OUTPUT_MAX_CHARS:
|
|
clipped = (
|
|
f"... omitted {len(clipped) - _TERMINAL_EXIT_OUTPUT_MAX_CHARS} "
|
|
"earlier character(s) ...\n"
|
|
f"{clipped[-_TERMINAL_EXIT_OUTPUT_MAX_CHARS:]}"
|
|
)
|
|
return clipped
|
|
|
|
|
|
def _terminal_exit_diagnostics(
|
|
instance: Any | None,
|
|
) -> tuple[str | None, int | None, str | None, str | None]:
|
|
"""Extract generic launch/output diagnostics from a terminal instance."""
|
|
if instance is None:
|
|
return None, None, None, None
|
|
|
|
raw_command = getattr(instance, "command", None)
|
|
command = raw_command if isinstance(raw_command, str) and raw_command else None
|
|
|
|
raw_args = getattr(instance, "args", None)
|
|
args_count = len(raw_args) if isinstance(raw_args, list) else None
|
|
|
|
raw_cwd = getattr(instance, "launch_cwd", None)
|
|
cwd = raw_cwd if isinstance(raw_cwd, str) and raw_cwd else None
|
|
|
|
last_output: str | None = None
|
|
read_last_output = getattr(instance, "last_pane_text", None)
|
|
if callable(read_last_output):
|
|
try:
|
|
raw_last_output = read_last_output()
|
|
except Exception:
|
|
_logger.exception("Failed to read terminal pane diagnostics")
|
|
else:
|
|
if isinstance(raw_last_output, str):
|
|
last_output = _trim_terminal_exit_output(raw_last_output)
|
|
|
|
return command, args_count, cwd, last_output
|
|
|
|
|
|
def _monotonic() -> float:
|
|
"""Return a monotonic timestamp for activity-emit throttling.
|
|
|
|
Thin indirection over :func:`time.monotonic` so tests can patch this
|
|
module-local symbol (per the project's mock-integrity guidance)
|
|
instead of mutating the process-wide ``time`` module.
|
|
|
|
:returns: Seconds from an arbitrary monotonic reference point.
|
|
"""
|
|
return time.monotonic()
|
|
|
|
|
|
def _sanitize_session_id(session_id: str) -> str:
|
|
"""Sanitize a session id for safe use as a filesystem path component.
|
|
|
|
:param session_id: Raw session/conversation identifier,
|
|
e.g. ``"conv_abc123"`` or ``"user/session"``.
|
|
:returns: Sanitized string safe for directory names.
|
|
"""
|
|
return session_id.replace("/", "_").replace("..", "_")
|
|
|
|
|
|
def _session_workspace(session_id: str) -> str:
|
|
"""Compute the workspace root for a session.
|
|
|
|
:param session_id: Session/conversation identifier,
|
|
e.g. ``"conv_abc123"``.
|
|
:returns: Absolute path to the session workspace directory.
|
|
"""
|
|
root = os.environ.get(
|
|
"OMNIGENT_RUNNER_OS_ENV_ROOT",
|
|
_DEFAULT_WORKSPACE_ROOT,
|
|
)
|
|
return os.path.join(root, _sanitize_session_id(session_id), "workspace")
|
|
|
|
|
|
class SessionResourceRegistry:
|
|
"""Runner-side registry that owns session-scoped resources.
|
|
|
|
Wraps :class:`TerminalRegistry` for terminal resources and
|
|
manages the primary :class:`OSEnvironment` per session. The
|
|
primary environment is lazily materialized on first
|
|
:meth:`resolve_environment` call with
|
|
``DEFAULT_ENVIRONMENT_ID``.
|
|
|
|
:param terminal_registry: Existing terminal registry.
|
|
``None`` in test setups without terminals.
|
|
:param runner_workspace: Optional CLI launch workspace. When set,
|
|
sessions whose spec cwd is unset or a placeholder (``"."``)
|
|
use this path instead of a per-session temp directory, keeping
|
|
the OS environment cwd aligned with the filesystem-registry
|
|
watch path.
|
|
:param per_session_workspace: When ``True`` and *runner_workspace*
|
|
is set, each session gets an isolated subdirectory under
|
|
*runner_workspace* instead of sharing the root.
|
|
Out-of-process (shared) runners should set this to ``True``;
|
|
in-process single-user runners leave it ``False`` so the agent
|
|
sees the project root directly.
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
terminal_registry: TerminalRegistry | None = None,
|
|
runner_workspace: Path | None = None,
|
|
*,
|
|
per_session_workspace: bool = False,
|
|
) -> None:
|
|
self._terminal_registry = terminal_registry
|
|
self._runner_workspace = runner_workspace
|
|
self._per_session_workspace = per_session_workspace
|
|
self._primary_envs: dict[str, OSEnvironment] = {}
|
|
self._terminal_roles: dict[tuple[str, str], str] = {}
|
|
self._terminal_lifecycles: dict[tuple[str, str], TerminalLifecycle] = {}
|
|
self._is_alive_cache: TTLCache[str, bool] = TTLCache(
|
|
maxsize=_IS_ALIVE_CACHE_MAX,
|
|
ttl=_IS_ALIVE_CACHE_TTL_S,
|
|
)
|
|
self._lock = threading.Lock()
|
|
# Optional callback ``(session_id, terminal_id) -> None`` invoked
|
|
# (on the event loop) when a terminal's pane produces output, so
|
|
# the runner can emit a ``session.terminal.activity`` SSE event.
|
|
# Set by the runner via :meth:`set_terminal_activity_publisher`.
|
|
self._terminal_activity_publisher: Callable[[str, str], None] | None = None
|
|
# Optional callback ``(session_id, status) -> None`` invoked (on
|
|
# the event loop) when the claude-native *agent* terminal's pane
|
|
# crosses an activity/idle edge, so the runner can emit a
|
|
# ``session.status`` event. This is the PTY-activity-derived
|
|
# working status that replaces the hook-based ``UserPromptSubmit``
|
|
# → running / ``Stop`` → idle bracketing. Set by the runner via
|
|
# :meth:`set_session_status_publisher`.
|
|
self._session_status_publisher: Callable[[str, str], None] | None = None
|
|
# Latest PTY-derived status (running/idle) per session. Lets
|
|
# :meth:`_handle_terminal_exit` tell a clean shutdown (idle) from a
|
|
# mid-turn crash. Written from the watcher thread and the turn-start
|
|
# hook; all access goes through the ``_*_session_status_memo`` helpers
|
|
# under ``self._lock``.
|
|
self._last_session_status: dict[str, str] = {}
|
|
# Optional callback invoked on the event loop when a watched terminal
|
|
# disappears unexpectedly. The callback receives the terminal's
|
|
# lifecycle relationship so the runner can decide whether the owning
|
|
# session should fail.
|
|
self._terminal_exit_publisher: Callable[[TerminalExitEvent], None] | None = None
|
|
|
|
def set_terminal_activity_publisher(
|
|
self,
|
|
publisher: Callable[[str, str], None],
|
|
) -> None:
|
|
"""Install the terminal-activity publisher.
|
|
|
|
The runner passes a callback that publishes a
|
|
``session.terminal.activity`` event onto the session's SSE
|
|
queue. It is invoked on the event loop (the watcher thread hops
|
|
via ``loop.call_soon_threadsafe``), so the callback itself may
|
|
use the loop-only ``_publish_event`` directly.
|
|
|
|
:param publisher: Callable ``(session_id, terminal_id) -> None``.
|
|
"""
|
|
self._terminal_activity_publisher = publisher
|
|
|
|
def set_session_status_publisher(
|
|
self,
|
|
publisher: Callable[[str, str], None],
|
|
) -> None:
|
|
"""Install the PTY-activity-derived session-status publisher.
|
|
|
|
The runner passes a callback that publishes a ``session.status``
|
|
event onto the session's SSE queue (which the Omnigent server relays
|
|
through its normal status path). It is invoked on the event loop
|
|
(the watcher thread hops via ``loop.call_soon_threadsafe``), so
|
|
the callback itself may use the loop-only ``_publish_event``
|
|
directly. Only the claude-native agent terminal's watcher calls
|
|
it — see :meth:`_start_terminal_activity_watcher`.
|
|
|
|
:param publisher: Callable ``(session_id, status) -> None`` where
|
|
*status* is ``"running"`` or ``"idle"``.
|
|
"""
|
|
self._session_status_publisher = publisher
|
|
|
|
def set_terminal_exit_publisher(
|
|
self,
|
|
publisher: Callable[[TerminalExitEvent], None],
|
|
) -> None:
|
|
"""Install the terminal-exit publisher.
|
|
|
|
The runner passes a callback that publishes resource/session lifecycle
|
|
events when a watched terminal disappears. It is invoked on the event
|
|
loop, not on the watcher thread.
|
|
|
|
Mental model:
|
|
required terminal:
|
|
If this terminal dies, the session is dead.
|
|
|
|
auxiliary terminal:
|
|
If this terminal dies, only this terminal resource is gone.
|
|
|
|
:param publisher: Callable receiving a :class:`TerminalExitEvent`.
|
|
"""
|
|
self._terminal_exit_publisher = publisher
|
|
|
|
def _set_session_status_memo(self, session_id: str, status: str) -> None:
|
|
"""Record the session's latest PTY status for exit classification."""
|
|
with self._lock:
|
|
self._last_session_status[session_id] = status
|
|
|
|
def _take_session_status_memo(self, session_id: str) -> str | None:
|
|
"""Pop and return the session's recorded PTY status (or ``None``)."""
|
|
with self._lock:
|
|
return self._last_session_status.pop(session_id, None)
|
|
|
|
def note_session_turn_started(self, session_id: str) -> None:
|
|
"""Mark a session as having an in-flight turn.
|
|
|
|
Closes the window between a new turn starting and the watcher's first
|
|
``running`` edge: without it, a crash in that gap would read the prior
|
|
turn's stale ``idle`` and be misclassified as a clean shutdown. The
|
|
watcher flips the memo back to ``idle`` once the turn completes.
|
|
|
|
:param session_id: Session/conversation identifier, e.g. ``"conv_abc"``.
|
|
"""
|
|
self._set_session_status_memo(session_id, "running")
|
|
|
|
def note_external_session_status(self, session_id: str, status: str) -> None:
|
|
"""Record a terminal-observed external status for exit classification.
|
|
|
|
Structured native forwarders can know turn completion more reliably than
|
|
a PTY diff heuristic. Keep the required-terminal exit memo aligned so a
|
|
terminal that closes after a forwarded ``idle`` edge is treated as a
|
|
clean shutdown, while ``running`` / ``waiting`` still classify a later
|
|
exit as mid-turn.
|
|
|
|
:param session_id: Session/conversation identifier, e.g. ``"conv_abc"``.
|
|
:param status: External native status, e.g. ``"running"`` or ``"idle"``.
|
|
"""
|
|
if status == "idle":
|
|
self._set_session_status_memo(session_id, "idle")
|
|
elif status in {"running", "waiting"}:
|
|
self._set_session_status_memo(session_id, "running")
|
|
|
|
@property
|
|
def terminal_registry(self) -> TerminalRegistry | None:
|
|
"""The wrapped terminal registry."""
|
|
return self._terminal_registry
|
|
|
|
def terminal_resource_role(
|
|
self,
|
|
session_id: str,
|
|
terminal_id: str,
|
|
) -> str | None:
|
|
"""Return the internal role marker for a terminal resource.
|
|
|
|
Role markers are runner-private state used to distinguish
|
|
runner-owned native terminals from generic terminals with the same
|
|
public id. They are intentionally not projected into public resource
|
|
metadata.
|
|
|
|
:param session_id: Session/conversation identifier, e.g.
|
|
``"conv_abc123"``.
|
|
:param terminal_id: Opaque terminal resource id, e.g.
|
|
``"terminal_codex_main"``.
|
|
:returns: Role marker such as ``"codex-native"``, or ``None``.
|
|
"""
|
|
with self._lock:
|
|
return self._terminal_roles.get((session_id, terminal_id))
|
|
|
|
def list_resources(
|
|
self,
|
|
session_id: str,
|
|
*,
|
|
resource_type: Literal["environment", "terminal", "file"] | None = None,
|
|
agent_spec: Any | None = None,
|
|
) -> PagedList[SessionResourceView]:
|
|
"""List all resources for a session.
|
|
|
|
Includes the logical default environment, running terminals,
|
|
and terminal-specific environments. When *agent_spec* is
|
|
provided and has no ``os_env``, the default environment
|
|
resource is omitted from the listing.
|
|
|
|
:param session_id: Session/conversation identifier.
|
|
:param resource_type: Optional filter by resource type.
|
|
:param agent_spec: Optional agent spec. When provided and
|
|
``agent_spec.os_env`` is ``None``, the logical default
|
|
environment resource is suppressed. ``None`` (the default)
|
|
preserves legacy behaviour and always includes the default
|
|
environment.
|
|
:returns: Paginated list of session resources.
|
|
"""
|
|
primary_os_env_spec = (
|
|
getattr(agent_spec, "os_env", None) if agent_spec is not None else None
|
|
)
|
|
has_os_env = agent_spec is None or primary_os_env_spec is not None
|
|
page = list_session_resources_from_terminal_registry(
|
|
session_id,
|
|
self._terminal_registry,
|
|
has_os_env=has_os_env,
|
|
primary_os_env_spec=primary_os_env_spec,
|
|
)
|
|
if resource_type is not None:
|
|
return filter_resources_by_type(page, resource_type)
|
|
return page
|
|
|
|
def get_resource(
|
|
self,
|
|
session_id: str,
|
|
resource_id: str,
|
|
) -> SessionResourceView | None:
|
|
"""Find a single resource by id.
|
|
|
|
:param session_id: Session/conversation identifier.
|
|
:param resource_id: Opaque resource id,
|
|
e.g. ``"default"`` or ``"terminal_bash_s1"``.
|
|
:returns: The matching resource or ``None``.
|
|
"""
|
|
page = list_session_resources_from_terminal_registry(
|
|
session_id,
|
|
self._terminal_registry,
|
|
)
|
|
return get_resource_by_id(page, resource_id)
|
|
|
|
async def get_terminal_resource(
|
|
self,
|
|
session_id: str,
|
|
terminal_id: str,
|
|
) -> SessionResourceView | None:
|
|
"""
|
|
Return a terminal resource after verifying tmux is still alive.
|
|
|
|
``TerminalInstance.running`` is an optimistic in-memory flag.
|
|
A terminal command can exit and take down the tmux server before
|
|
any send/read/close path updates that flag. Terminal GET uses
|
|
this method so clients do not reconnect to a stale socket that
|
|
can only print tmux's ``"no sessions"`` error.
|
|
|
|
The ``is_alive()`` subprocess probe is cached for a short TTL
|
|
so rapid polling from web clients does not fork a ``tmux
|
|
has-session`` process on every request.
|
|
|
|
:param session_id: Session/conversation identifier.
|
|
:param terminal_id: Opaque terminal resource id,
|
|
e.g. ``"terminal_claude_main"``.
|
|
:returns: A terminal resource view when the matching tmux
|
|
server is alive; otherwise ``None``.
|
|
"""
|
|
if self._terminal_registry is None:
|
|
return None
|
|
|
|
for entry in self._terminal_registry.list_for_conversation(
|
|
session_id,
|
|
):
|
|
if terminal_resource_id(entry.terminal_name, entry.session_key) != terminal_id:
|
|
continue
|
|
if not entry.instance.running:
|
|
return None
|
|
|
|
cache_key = f"{session_id}:{terminal_id}"
|
|
cached = self._is_alive_cache.get(cache_key)
|
|
if cached is not None:
|
|
return terminal_resource_view(session_id, entry) if cached else None
|
|
|
|
alive = await entry.instance.is_alive()
|
|
if alive:
|
|
self._is_alive_cache[cache_key] = True
|
|
else:
|
|
self._is_alive_cache.pop(cache_key, None)
|
|
return None
|
|
return terminal_resource_view(session_id, entry)
|
|
return None
|
|
|
|
def resolve_environment(
|
|
self,
|
|
session_id: str,
|
|
environment_id: str,
|
|
agent_spec: Any | None = None,
|
|
) -> OSEnvironment:
|
|
"""Resolve an environment id to a live OSEnvironment.
|
|
|
|
For ``DEFAULT_ENVIRONMENT_ID``, lazily creates the primary
|
|
environment from the agent spec (or synthesizes a default).
|
|
For terminal environment ids, resolves from the terminal
|
|
registry.
|
|
|
|
:param session_id: Session/conversation identifier.
|
|
:param environment_id: Environment resource id.
|
|
:param agent_spec: Agent spec for primary env creation.
|
|
:returns: The live :class:`OSEnvironment`.
|
|
:raises ValueError: If the environment id cannot be resolved.
|
|
"""
|
|
if environment_id == DEFAULT_ENVIRONMENT_ID:
|
|
return self._resolve_primary(session_id, agent_spec)
|
|
|
|
if self._terminal_registry is not None:
|
|
for entry in self._terminal_registry.list_for_conversation(
|
|
session_id,
|
|
):
|
|
if not entry.instance.running:
|
|
continue
|
|
env_id = terminal_environment_resource_id(
|
|
entry.terminal_name,
|
|
entry.session_key,
|
|
)
|
|
if env_id == environment_id and entry.instance.os_env is not None:
|
|
return entry.instance.os_env
|
|
|
|
raise ValueError(f"Environment {environment_id!r} not found for session {session_id!r}")
|
|
|
|
def _resolve_primary(
|
|
self,
|
|
session_id: str,
|
|
agent_spec: Any | None,
|
|
) -> OSEnvironment:
|
|
"""Get or create the primary OSEnvironment for a session.
|
|
|
|
:param session_id: Session/conversation identifier.
|
|
:param agent_spec: Agent spec for env creation.
|
|
:returns: The primary :class:`OSEnvironment`.
|
|
"""
|
|
with self._lock:
|
|
cached = self._primary_envs.get(session_id)
|
|
if cached is not None:
|
|
return cached
|
|
|
|
os_env = self._create_primary_env(session_id, agent_spec)
|
|
self._primary_envs[session_id] = os_env
|
|
return os_env
|
|
|
|
def _create_primary_env(
|
|
self,
|
|
session_id: str,
|
|
agent_spec: Any | None,
|
|
) -> OSEnvironment:
|
|
"""Create a new primary OSEnvironment.
|
|
|
|
Follows the creation policy from the design:
|
|
1. If agent_spec.os_env exists, clone it
|
|
2. Resolve cwd to session workspace if unset
|
|
3. If agent_spec is None, synthesize a default spec
|
|
|
|
The default branch (no agent_spec) serves the
|
|
filesystem-resource endpoints — a read view, never agent
|
|
tool execution. Pin ``sandbox.type="none"`` so it can't
|
|
inherit the Linux platform default (bwrap), which
|
|
raises when the ``bwrap`` binary is missing and broke the
|
|
working-folder panel for runners without it.
|
|
|
|
:param session_id: Session/conversation identifier.
|
|
:param agent_spec: Agent spec for env creation.
|
|
:returns: The newly created :class:`OSEnvironment`.
|
|
:raises ValueError: If ``agent_spec`` is provided but its
|
|
``os_env`` field is ``None``. Callers must gate on
|
|
``os_env`` presence before materialising an environment.
|
|
"""
|
|
from omnigent.inner.datamodel import OSEnvSandboxSpec, OSEnvSpec
|
|
from omnigent.inner.os_env import create_os_environment
|
|
|
|
# Prefer the CLI launch workspace so that the OS environment
|
|
# cwd matches the filesystem-registry watch path. Fall back
|
|
# to the per-session temp dir for remote/cloud runners that
|
|
# have no workspace affinity.
|
|
if self._runner_workspace is not None:
|
|
if self._per_session_workspace:
|
|
# Isolate sessions under the shared workspace.
|
|
default_cwd = str(self._runner_workspace / _sanitize_session_id(session_id))
|
|
os.makedirs(default_cwd, mode=0o700, exist_ok=True)
|
|
os.chmod(default_cwd, 0o700) # ensure mode even if pre-existing
|
|
else:
|
|
default_cwd = str(self._runner_workspace)
|
|
else:
|
|
default_cwd = _session_workspace(session_id)
|
|
# Restrict workspace visibility on shared hosts.
|
|
os.makedirs(default_cwd, mode=0o700, exist_ok=True)
|
|
os.chmod(default_cwd, 0o700) # ensure mode even if pre-existing
|
|
|
|
if agent_spec is not None:
|
|
spec_os_env = getattr(agent_spec, "os_env", None)
|
|
if spec_os_env is None:
|
|
raise ValueError(
|
|
"Agent spec has no os_env; cannot create a primary filesystem environment."
|
|
)
|
|
# Precedence per designs/SESSION_WORKSPACE_SELECTION.md:
|
|
# runner_workspace (env-var-driven) ALWAYS wins when set.
|
|
# Otherwise the spec's absolute cwd wins; otherwise we
|
|
# fall back to the per-session tmpdir (default_cwd).
|
|
if (
|
|
self._runner_workspace is not None
|
|
or spec_os_env.cwd is None
|
|
or spec_os_env.cwd in (".", "./")
|
|
):
|
|
cwd = default_cwd
|
|
else:
|
|
cwd = spec_os_env.cwd
|
|
effective_spec = OSEnvSpec(
|
|
type=spec_os_env.type,
|
|
cwd=cwd,
|
|
sandbox=spec_os_env.sandbox,
|
|
fork=spec_os_env.fork,
|
|
start_in_scratch=spec_os_env.start_in_scratch,
|
|
)
|
|
env = create_os_environment(effective_spec)
|
|
if env is not None:
|
|
return env
|
|
|
|
default_spec = OSEnvSpec(
|
|
type="caller_process",
|
|
cwd=default_cwd,
|
|
sandbox=OSEnvSandboxSpec(type="none"),
|
|
)
|
|
env = create_os_environment(default_spec)
|
|
if env is None:
|
|
raise RuntimeError(
|
|
f"Failed to create default OS environment for session {session_id!r}"
|
|
)
|
|
return env
|
|
|
|
def compute_default_env_root(
|
|
self,
|
|
session_id: str,
|
|
agent_spec: Any | None,
|
|
) -> str | None:
|
|
"""Compute the resolved filesystem root for the default environment.
|
|
|
|
Mirrors the cwd resolution logic in :meth:`_create_primary_env` without
|
|
materializing the :class:`OSEnvironment`. Safe to call from listing and
|
|
single-fetch endpoints that must remain logical/lazy.
|
|
|
|
Precedence (per
|
|
designs/SESSION_WORKSPACE_SELECTION.md "How this maps onto runtime"):
|
|
|
|
1. ``self._runner_workspace`` (sourced from
|
|
``OMNIGENT_RUNNER_WORKSPACE``) — when set, ALWAYS
|
|
wins. Both CLI- and host-launched sessions populate it
|
|
with the authoritative starting cwd; any spec cwd is
|
|
treated as a session-create-time boundary, not a
|
|
runtime override.
|
|
2. The agent's absolute ``os_env.cwd``, when present and
|
|
not a relative placeholder. Used for pure local runs
|
|
that bypass the env-var path (e.g. unit tests that
|
|
construct an ``AgentSpec`` directly).
|
|
3. The per-session default workspace tmpdir.
|
|
|
|
:param session_id: Session/conversation identifier.
|
|
:param agent_spec: Agent spec for the session. When provided and its
|
|
``os_env`` field is ``None``, the session has no filesystem and
|
|
``None`` is returned. When ``None`` (dev/standalone mode) the
|
|
default workspace path is returned.
|
|
:returns: Resolved absolute root path string, or ``None`` when the
|
|
session has no filesystem.
|
|
"""
|
|
# No-os_env agents have no filesystem regardless of how the
|
|
# runner was launched — keep that signal so the route layer
|
|
# 404s on filesystem endpoints for headless agents.
|
|
if agent_spec is not None:
|
|
spec_os_env = getattr(agent_spec, "os_env", None)
|
|
if spec_os_env is None:
|
|
return None
|
|
|
|
# Runner workspace wins when set. Per-session subdirectory
|
|
# isolation is preserved so concurrent sessions
|
|
# don't share a cwd.
|
|
if self._runner_workspace is not None:
|
|
if self._per_session_workspace:
|
|
default_cwd = str(self._runner_workspace / _sanitize_session_id(session_id))
|
|
else:
|
|
default_cwd = str(self._runner_workspace)
|
|
return str(Path(default_cwd).resolve())
|
|
|
|
# No runner workspace → fall back to the spec's cwd if it's
|
|
# a real absolute path, otherwise the per-session tmpdir.
|
|
if agent_spec is not None:
|
|
spec_os_env = getattr(agent_spec, "os_env", None)
|
|
cwd = getattr(spec_os_env, "cwd", None) if spec_os_env is not None else None
|
|
if cwd is not None and cwd not in (".", "./"):
|
|
return str(Path(cwd).resolve())
|
|
|
|
# Last resort: compute path without os.makedirs — read-only computation.
|
|
default_cwd = _session_workspace(session_id)
|
|
return str(Path(default_cwd).resolve())
|
|
|
|
async def launch_required_terminal(
|
|
self,
|
|
session_id: str,
|
|
terminal_name: str,
|
|
session_key: str,
|
|
spec: Any,
|
|
*,
|
|
cwd_override: str | None = None,
|
|
sandbox_override: str | None = None,
|
|
parent_os_env: Any | None = None,
|
|
resource_role: str | None = None,
|
|
) -> SessionResourceView:
|
|
"""Launch a terminal required for the owning session to execute.
|
|
|
|
Mental model:
|
|
If this terminal dies, the session is dead.
|
|
|
|
Use this when the terminal process is the session runtime, or an
|
|
essential part of that runtime. This is agent-independent: callers
|
|
declare a lifecycle relationship, not a vendor or harness type.
|
|
"""
|
|
return await self._launch_terminal_with_lifecycle(
|
|
TerminalLifecycle.REQUIRED,
|
|
session_id=session_id,
|
|
terminal_name=terminal_name,
|
|
session_key=session_key,
|
|
spec=spec,
|
|
cwd_override=cwd_override,
|
|
sandbox_override=sandbox_override,
|
|
parent_os_env=parent_os_env,
|
|
resource_role=resource_role,
|
|
)
|
|
|
|
async def launch_auxiliary_terminal(
|
|
self,
|
|
session_id: str,
|
|
terminal_name: str,
|
|
session_key: str,
|
|
spec: Any,
|
|
*,
|
|
cwd_override: str | None = None,
|
|
sandbox_override: str | None = None,
|
|
parent_os_env: Any | None = None,
|
|
resource_role: str | None = None,
|
|
) -> SessionResourceView:
|
|
"""Launch a terminal resource attached to the owning session.
|
|
|
|
Mental model:
|
|
If this terminal dies, only this terminal resource is gone.
|
|
|
|
Use this for terminals that provide UI access, logs, debugging, REPLs,
|
|
or optional interaction. This is agent-independent: callers declare a
|
|
lifecycle relationship, not a vendor or harness type.
|
|
"""
|
|
return await self._launch_terminal_with_lifecycle(
|
|
TerminalLifecycle.AUXILIARY,
|
|
session_id=session_id,
|
|
terminal_name=terminal_name,
|
|
session_key=session_key,
|
|
spec=spec,
|
|
cwd_override=cwd_override,
|
|
sandbox_override=sandbox_override,
|
|
parent_os_env=parent_os_env,
|
|
resource_role=resource_role,
|
|
)
|
|
|
|
async def _launch_terminal_with_lifecycle(
|
|
self,
|
|
lifecycle: TerminalLifecycle,
|
|
*,
|
|
session_id: str,
|
|
terminal_name: str,
|
|
session_key: str,
|
|
spec: Any,
|
|
cwd_override: str | None = None,
|
|
sandbox_override: str | None = None,
|
|
parent_os_env: Any | None = None,
|
|
resource_role: str | None = None,
|
|
) -> SessionResourceView:
|
|
"""Launch a terminal, then observe it with the requested lifecycle."""
|
|
if self._terminal_registry is None:
|
|
raise RuntimeError("Terminal registry not configured")
|
|
|
|
instance = await self._terminal_registry.launch(
|
|
conversation_id=session_id,
|
|
terminal_name=terminal_name,
|
|
session_key=session_key,
|
|
spec=spec,
|
|
parent_os_env=parent_os_env,
|
|
cwd_override=cwd_override,
|
|
sandbox_override=sandbox_override,
|
|
)
|
|
return await self._observe_terminal_with_lifecycle(
|
|
lifecycle,
|
|
session_id=session_id,
|
|
terminal_name=terminal_name,
|
|
session_key=session_key,
|
|
instance=instance,
|
|
resource_role=resource_role,
|
|
)
|
|
|
|
async def observe_required_terminal(
|
|
self,
|
|
session_id: str,
|
|
terminal_name: str,
|
|
session_key: str,
|
|
instance: Any,
|
|
*,
|
|
resource_role: str | None = None,
|
|
) -> SessionResourceView:
|
|
"""Observe an existing terminal required for session execution.
|
|
|
|
Mental model:
|
|
If this terminal dies, the session is dead.
|
|
"""
|
|
return await self._observe_terminal_with_lifecycle(
|
|
TerminalLifecycle.REQUIRED,
|
|
session_id=session_id,
|
|
terminal_name=terminal_name,
|
|
session_key=session_key,
|
|
instance=instance,
|
|
resource_role=resource_role,
|
|
)
|
|
|
|
async def observe_auxiliary_terminal(
|
|
self,
|
|
session_id: str,
|
|
terminal_name: str,
|
|
session_key: str,
|
|
instance: Any,
|
|
*,
|
|
resource_role: str | None = None,
|
|
) -> SessionResourceView:
|
|
"""Observe an existing terminal attached to the owning session.
|
|
|
|
Mental model:
|
|
If this terminal dies, only this terminal resource is gone.
|
|
"""
|
|
return await self._observe_terminal_with_lifecycle(
|
|
TerminalLifecycle.AUXILIARY,
|
|
session_id=session_id,
|
|
terminal_name=terminal_name,
|
|
session_key=session_key,
|
|
instance=instance,
|
|
resource_role=resource_role,
|
|
)
|
|
|
|
async def _observe_terminal_with_lifecycle(
|
|
self,
|
|
lifecycle: TerminalLifecycle,
|
|
*,
|
|
session_id: str,
|
|
terminal_name: str,
|
|
session_key: str,
|
|
instance: Any,
|
|
resource_role: str | None = None,
|
|
) -> SessionResourceView:
|
|
"""Project and observe an already-launched terminal instance."""
|
|
if self._terminal_registry is None:
|
|
raise RuntimeError("Terminal registry not configured")
|
|
if not getattr(instance, "running", False) or not await instance.is_alive():
|
|
await self._terminal_registry.close(session_id, terminal_name, session_key)
|
|
raise RuntimeError(
|
|
f"terminal {terminal_name}:{session_key} is not running for session {session_id}"
|
|
)
|
|
|
|
from omnigent.terminals.registry import TerminalListEntry
|
|
|
|
resource_id = terminal_resource_id(terminal_name, session_key)
|
|
with self._lock:
|
|
previous = self._terminal_lifecycles.get((session_id, resource_id))
|
|
if previous is not None and previous != lifecycle:
|
|
raise RuntimeError(
|
|
f"terminal {terminal_name}:{session_key} for session {session_id} "
|
|
f"is already observed as {previous.value}"
|
|
)
|
|
self._terminal_lifecycles[(session_id, resource_id)] = lifecycle
|
|
if resource_role is not None:
|
|
self._terminal_roles[(session_id, resource_id)] = resource_role
|
|
self._start_terminal_activity_watcher(
|
|
session_id,
|
|
terminal_name,
|
|
session_key,
|
|
instance,
|
|
resource_role,
|
|
lifecycle,
|
|
replace=True,
|
|
)
|
|
return terminal_resource_view(
|
|
session_id,
|
|
TerminalListEntry(
|
|
terminal_name=terminal_name,
|
|
session_key=session_key,
|
|
instance=instance,
|
|
),
|
|
)
|
|
|
|
def _start_terminal_activity_watcher(
|
|
self,
|
|
session_id: str,
|
|
terminal_name: str,
|
|
session_key: str,
|
|
instance: Any,
|
|
resource_role: str | None,
|
|
lifecycle: TerminalLifecycle,
|
|
*,
|
|
replace: bool = False,
|
|
) -> None:
|
|
"""Start (idempotently) the per-terminal pane-activity watcher.
|
|
|
|
Drives the runner-determined "PTY had output" signal that powers
|
|
the web terminal-activity badge, replacing the removed
|
|
per-terminal client WS attach. The same watcher also reports
|
|
unexpected terminal exit so resource/session lifecycle stays
|
|
aligned with the underlying tmux process. No-op when no publisher
|
|
is installed (e.g. embedded/test runners).
|
|
|
|
For the claude-native *agent* terminal (``resource_role`` ==
|
|
:data:`CLAUDE_NATIVE_TERMINAL_ROLE` or
|
|
:data:`PI_NATIVE_TERMINAL_ROLE`) the same watcher also drives the
|
|
session's working status: pane activity → ``running`` and a short
|
|
quiescence → ``idle``, emitted via the session-status publisher.
|
|
This PTY-derived status catches cases lifecycle hooks can miss
|
|
because it observes the terminal directly. The status edges are
|
|
gated to these roles so a side shell's output never flips the
|
|
session's status.
|
|
|
|
:param session_id: Session/conversation identifier.
|
|
:param terminal_name: Terminal name from the agent spec.
|
|
:param session_key: Per-launch session key.
|
|
:param instance: The launched :class:`TerminalInstance`.
|
|
:param resource_role: Runner-private role marker for this
|
|
terminal, e.g. :data:`CLAUDE_NATIVE_TERMINAL_ROLE`, or
|
|
``None`` for a generic terminal (activity badge only).
|
|
:param lifecycle: Required/auxiliary relationship between this
|
|
terminal and the owning session.
|
|
:param replace: Whether to replace an existing watcher so callbacks
|
|
can be rebound after terminal ownership transfer.
|
|
"""
|
|
activity_publisher = self._terminal_activity_publisher
|
|
status_publisher = self._session_status_publisher
|
|
exit_publisher = self._terminal_exit_publisher
|
|
# Status edges are derived only from native agent terminals — a
|
|
# generic shell's output must not move the session's working status.
|
|
emit_status = status_publisher is not None and resource_role in {
|
|
CLAUDE_NATIVE_TERMINAL_ROLE,
|
|
PI_NATIVE_TERMINAL_ROLE,
|
|
# cursor-native has no forwarder/hook (run_turn returns immediately
|
|
# after the paste), so — like pi/claude — the PTY watcher is its only
|
|
# status source. Without this the web "Working…" badge never clears.
|
|
CURSOR_NATIVE_TERMINAL_ROLE,
|
|
KIRO_NATIVE_TERMINAL_ROLE,
|
|
# goose-native injects then returns (its forwarder only mirrors the
|
|
# transcript, not status), so the PTY watcher is its status source too.
|
|
GOOSE_NATIVE_TERMINAL_ROLE,
|
|
# qwen-native appends then returns (its forwarder only mirrors the
|
|
# JSON event transcript, not status), so the PTY watcher is its
|
|
# status source too.
|
|
QWEN_NATIVE_TERMINAL_ROLE,
|
|
# kimi-native also has no forwarder/hook (the injection run_turn
|
|
# returns right after the tmux paste), so the PTY watcher is its
|
|
# only running/idle status source — same as cursor/pi/claude.
|
|
KIMI_NATIVE_TERMINAL_ROLE,
|
|
# hermes-native injects then returns (its forwarder only mirrors the
|
|
# SQLite transcript, not status), so the PTY watcher is its status
|
|
# source too.
|
|
HERMES_NATIVE_TERMINAL_ROLE,
|
|
}
|
|
if activity_publisher is None and not emit_status and exit_publisher is None:
|
|
return
|
|
resource_id = terminal_resource_id(terminal_name, session_key)
|
|
loop = asyncio.get_running_loop()
|
|
|
|
# Last status edge emitted for this terminal, mutated only on the
|
|
# watcher daemon thread (single-threaded), so the running/idle
|
|
# transition is deduped without a lock. Scoped to this watcher's
|
|
# lifetime, so a fresh terminal gets a fresh baseline — no stale
|
|
# cross-launch state to clean up.
|
|
last_status: dict[str, str | None] = {"value": None}
|
|
# Monotonic time of the last activity pulse published for this
|
|
# terminal, mutated only on the watcher daemon thread (so no lock),
|
|
# used to throttle emissions to at most one per
|
|
# :data:`_TERMINAL_ACTIVITY_EMIT_MIN_INTERVAL_SECONDS`. ``None``
|
|
# means "never emitted", so the first changed tick always fires.
|
|
last_activity_emit: dict[str, float | None] = {"value": None}
|
|
|
|
def _on_activity() -> None:
|
|
# Runs on the watcher daemon thread; hop to the loop so the
|
|
# loop-only publishers (queue.put_nowait) are touched safely.
|
|
#
|
|
# Throttle the activity pulse to one per second: the
|
|
# claude-native pane changes on nearly every 200ms poll while
|
|
# Claude works, but the web badge only needs a pulse inside its
|
|
# 1.5s window — emitting on every tick would push ~5 events/sec
|
|
# of redundant traffic.
|
|
if activity_publisher is not None:
|
|
now = _monotonic()
|
|
previous = last_activity_emit["value"]
|
|
if (
|
|
previous is None
|
|
or now - previous >= _TERMINAL_ACTIVITY_EMIT_MIN_INTERVAL_SECONDS
|
|
):
|
|
last_activity_emit["value"] = now
|
|
loop.call_soon_threadsafe(activity_publisher, session_id, resource_id)
|
|
# Pane changed → the agent is working. Coalesce to the
|
|
# idle→running edge so a continuously-redrawing pane doesn't
|
|
# re-emit ``running`` every poll.
|
|
if emit_status and status_publisher is not None and last_status["value"] != "running":
|
|
last_status["value"] = "running"
|
|
# Track for exit classification: a pane exit now reads as a crash.
|
|
self._set_session_status_memo(session_id, "running")
|
|
loop.call_soon_threadsafe(status_publisher, session_id, "running")
|
|
|
|
def _on_exit() -> None:
|
|
def _schedule() -> None:
|
|
task = asyncio.create_task(
|
|
self._handle_terminal_exit(
|
|
session_id=session_id,
|
|
terminal_name=terminal_name,
|
|
session_key=session_key,
|
|
lifecycle=lifecycle,
|
|
instance=instance,
|
|
)
|
|
)
|
|
task.add_done_callback(_log_terminal_exit_task_result)
|
|
|
|
try:
|
|
loop.call_soon_threadsafe(_schedule)
|
|
except RuntimeError:
|
|
_logger.debug(
|
|
"Event loop unavailable while handling terminal exit: "
|
|
"session=%s terminal=%s:%s",
|
|
session_id,
|
|
terminal_name,
|
|
session_key,
|
|
)
|
|
|
|
def _log_terminal_exit_task_result(task: asyncio.Task[None]) -> None:
|
|
try:
|
|
task.result()
|
|
except asyncio.CancelledError:
|
|
return
|
|
except Exception:
|
|
_logger.exception(
|
|
"Terminal exit cleanup failed: session=%s terminal=%s:%s",
|
|
session_id,
|
|
terminal_name,
|
|
session_key,
|
|
)
|
|
|
|
if not emit_status:
|
|
instance.start_idle_watcher_thread(
|
|
on_activity=_on_activity if activity_publisher is not None else None,
|
|
on_exit=_on_exit,
|
|
replace=replace,
|
|
)
|
|
return
|
|
|
|
def _on_idle() -> None:
|
|
# Pane quiet for the claude-native status threshold → the
|
|
# agent has stopped. Edge-triggered: re-arms only after new
|
|
# output mutates the pane (which flips back to ``running``).
|
|
if status_publisher is not None and last_status["value"] != "idle":
|
|
last_status["value"] = "idle"
|
|
# Track for exit classification: a pane exit now reads as clean.
|
|
# Edge ordering: the watcher thread runs idle/exit serially, so
|
|
# this idle commits before any later on_exit reads the memo.
|
|
self._set_session_status_memo(session_id, "idle")
|
|
loop.call_soon_threadsafe(status_publisher, session_id, "idle")
|
|
# Clear the activity throttle so the next working episode emits
|
|
# its first pulse immediately, keeping the activity badge
|
|
# aligned with the running-status edge (which also re-fires on
|
|
# the next pane change) rather than lagging up to a second
|
|
# behind it.
|
|
last_activity_emit["value"] = None
|
|
|
|
instance.start_idle_watcher_thread(
|
|
on_activity=_on_activity,
|
|
on_idle=_on_idle,
|
|
on_exit=_on_exit,
|
|
idle_threshold_s=_CLAUDE_NATIVE_STATUS_IDLE_THRESHOLD_SECONDS,
|
|
poll_interval_s=_CLAUDE_NATIVE_STATUS_POLL_INTERVAL_SECONDS,
|
|
replace=replace,
|
|
)
|
|
|
|
async def _handle_terminal_exit(
|
|
self,
|
|
*,
|
|
session_id: str,
|
|
terminal_name: str,
|
|
session_key: str,
|
|
lifecycle: TerminalLifecycle,
|
|
instance: Any | None = None,
|
|
) -> None:
|
|
"""Clean up and publish lifecycle events for an unexpected terminal exit."""
|
|
terminal_id = terminal_resource_id(terminal_name, session_key)
|
|
with self._lock:
|
|
observed = self._terminal_lifecycles.pop((session_id, terminal_id), None)
|
|
self._terminal_roles.pop((session_id, terminal_id), None)
|
|
if observed is None:
|
|
return
|
|
if observed != lifecycle:
|
|
_logger.warning(
|
|
"Terminal lifecycle changed before exit handling: session=%s terminal=%s:%s "
|
|
"observed=%s callback=%s",
|
|
session_id,
|
|
terminal_name,
|
|
session_key,
|
|
observed.value,
|
|
lifecycle.value,
|
|
)
|
|
lifecycle = observed
|
|
|
|
command, args_count, cwd, last_output = _terminal_exit_diagnostics(instance)
|
|
# Idle = clean shutdown after the turn finished. Anything else (running,
|
|
# or never observed → boot failure) stays a failure.
|
|
session_was_idle = self._take_session_status_memo(session_id) == "idle"
|
|
|
|
if self._terminal_registry is not None:
|
|
try:
|
|
await self._terminal_registry.close(session_id, terminal_name, session_key)
|
|
except Exception:
|
|
_logger.exception(
|
|
"Error evicting exited terminal: session=%s terminal=%s:%s",
|
|
session_id,
|
|
terminal_name,
|
|
session_key,
|
|
)
|
|
|
|
publisher = self._terminal_exit_publisher
|
|
if publisher is not None:
|
|
publisher(
|
|
TerminalExitEvent(
|
|
session_id=session_id,
|
|
terminal_id=terminal_id,
|
|
terminal_name=terminal_name,
|
|
session_key=session_key,
|
|
lifecycle=lifecycle,
|
|
command=command,
|
|
args_count=args_count,
|
|
cwd=cwd,
|
|
last_output=last_output,
|
|
session_was_idle=session_was_idle,
|
|
)
|
|
)
|
|
|
|
async def close_terminal(
|
|
self,
|
|
session_id: str,
|
|
terminal_id: str,
|
|
) -> bool:
|
|
"""Close a terminal resource by id.
|
|
|
|
:param session_id: Session/conversation identifier.
|
|
:param terminal_id: Opaque terminal resource id.
|
|
:returns: ``True`` if a terminal was closed.
|
|
"""
|
|
if self._terminal_registry is None:
|
|
return False
|
|
|
|
for entry in self._terminal_registry.list_for_conversation(
|
|
session_id,
|
|
):
|
|
if terminal_resource_id(entry.terminal_name, entry.session_key) == terminal_id:
|
|
closed = await self._terminal_registry.close(
|
|
session_id,
|
|
entry.terminal_name,
|
|
entry.session_key,
|
|
)
|
|
if closed:
|
|
with self._lock:
|
|
self._terminal_roles.pop((session_id, terminal_id), None)
|
|
self._terminal_lifecycles.pop((session_id, terminal_id), None)
|
|
return closed
|
|
return False
|
|
|
|
async def transfer_terminal(
|
|
self,
|
|
source_session_id: str,
|
|
target_session_id: str,
|
|
terminal_id: str,
|
|
) -> SessionResourceView | None:
|
|
"""Move a terminal resource between sessions without closing it.
|
|
|
|
The underlying tmux pane remains live. Only the registry owner
|
|
key changes, and the returned resource is projected under the
|
|
target session id.
|
|
|
|
:param source_session_id: Current owning session id, e.g.
|
|
``"conv_old"``.
|
|
:param target_session_id: New owning session id, e.g.
|
|
``"conv_new"``.
|
|
:param terminal_id: Opaque terminal resource id, e.g.
|
|
``"terminal_claude_main"``.
|
|
:returns: The transferred terminal resource view under
|
|
*target_session_id*, or ``None`` if no matching source
|
|
terminal exists.
|
|
:raises RuntimeError: If the target session already has a
|
|
terminal with the same name and session key.
|
|
"""
|
|
if self._terminal_registry is None:
|
|
return None
|
|
|
|
from omnigent.terminals.registry import TerminalListEntry
|
|
|
|
for entry in self._terminal_registry.list_for_conversation(
|
|
source_session_id,
|
|
):
|
|
if not entry.instance.running:
|
|
continue
|
|
if terminal_resource_id(entry.terminal_name, entry.session_key) != terminal_id:
|
|
continue
|
|
moved = self._terminal_registry.transfer(
|
|
source_session_id,
|
|
target_session_id,
|
|
entry.terminal_name,
|
|
entry.session_key,
|
|
)
|
|
if not moved:
|
|
return None
|
|
with self._lock:
|
|
role = self._terminal_roles.pop((source_session_id, terminal_id), None)
|
|
if role is not None:
|
|
self._terminal_roles[(target_session_id, terminal_id)] = role
|
|
lifecycle = self._terminal_lifecycles.pop((source_session_id, terminal_id), None)
|
|
if lifecycle is not None:
|
|
self._terminal_lifecycles[(target_session_id, terminal_id)] = lifecycle
|
|
# Move the PTY-status memo with the pane so a post-transfer exit is
|
|
# classified against the right session. Don't clobber a status the
|
|
# target already has from its own terminal.
|
|
with self._lock:
|
|
moved_status = self._last_session_status.pop(source_session_id, None)
|
|
if moved_status is not None and target_session_id not in self._last_session_status:
|
|
self._last_session_status[target_session_id] = moved_status
|
|
try:
|
|
await entry.instance.set_conversation_link(
|
|
self._terminal_registry.conversation_link_for_id(target_session_id)
|
|
)
|
|
except (RuntimeError, OSError) as exc:
|
|
_logger.warning(
|
|
"Failed to update terminal status link after transfer to %s: %s",
|
|
target_session_id,
|
|
exc,
|
|
)
|
|
if lifecycle is not None:
|
|
self._start_terminal_activity_watcher(
|
|
target_session_id,
|
|
entry.terminal_name,
|
|
entry.session_key,
|
|
entry.instance,
|
|
role,
|
|
lifecycle,
|
|
replace=True,
|
|
)
|
|
return terminal_resource_view(
|
|
target_session_id,
|
|
TerminalListEntry(
|
|
terminal_name=entry.terminal_name,
|
|
session_key=entry.session_key,
|
|
instance=entry.instance,
|
|
),
|
|
)
|
|
return None
|
|
|
|
async def cleanup_session(self, session_id: str) -> None:
|
|
"""Close all resources owned by a session.
|
|
|
|
Closes the primary OSEnv and delegates terminal cleanup
|
|
to the terminal registry. Preserves workspace files for
|
|
post-mortem inspection per the design.
|
|
|
|
:param session_id: Session/conversation identifier.
|
|
"""
|
|
self._take_session_status_memo(session_id)
|
|
with self._lock:
|
|
primary = self._primary_envs.pop(session_id, None)
|
|
stale_role_keys = [key for key in self._terminal_roles if key[0] == session_id]
|
|
for key in stale_role_keys:
|
|
self._terminal_roles.pop(key, None)
|
|
stale_lifecycle_keys = [
|
|
key for key in self._terminal_lifecycles if key[0] == session_id
|
|
]
|
|
for key in stale_lifecycle_keys:
|
|
self._terminal_lifecycles.pop(key, None)
|
|
if primary is not None:
|
|
try:
|
|
primary.close()
|
|
except Exception:
|
|
_logger.exception(
|
|
"Error closing primary env for session=%s",
|
|
session_id,
|
|
)
|
|
|
|
if self._terminal_registry is not None:
|
|
try:
|
|
await self._terminal_registry.cleanup_conversation(
|
|
session_id,
|
|
)
|
|
except Exception:
|
|
_logger.exception(
|
|
"Error cleaning up terminals for session=%s",
|
|
session_id,
|
|
)
|
|
|
|
def has_primary_env(self, session_id: str) -> bool:
|
|
"""Check if a primary env has been materialized.
|
|
|
|
:param session_id: Session/conversation identifier.
|
|
:returns: ``True`` if the primary env is cached.
|
|
"""
|
|
with self._lock:
|
|
return session_id in self._primary_envs
|