core/tools: PTY tools use instance-aware manager facade

src/plyngent/tools/context.py

@@ -35,12 +35,18 @@ class InstanceState:
     # Ephemeral process maps (PTY etc.) may hang here later.
     extras: dict[str, Any] = field(default_factory=dict)
 
+    @property
+    def pty(self) -> Any:
+        """PTY manager facade for this process (class-level sessions today)."""
+        from plyngent.tools.process.pty_session import PtyManager
+
+        return PtyManager
+
     async def shutdown(self) -> None:
         """Best-effort cleanup hooks (PTY close, temp workspaces)."""
-        from plyngent.tools.process.pty_session import PtyManager
         from plyngent.tools.temp_workspace import cleanup_temporary_workspaces
 
-        PtyManager.close_all()
+        self.pty.close_all()
         _ = cleanup_temporary_workspaces()
 
 

src/plyngent/tools/process/ask_into_pty.py

@@ -6,7 +6,7 @@ from plyngent.agent import ToolTag, tool
 from plyngent.prompting import NonInteractiveError, ask_async, ask_secret_async
 from plyngent.tools.workspace import WorkspaceError
 
-from .pty_session import PtyManager
+from .pty_session import active_pty_manager
 from .write_pty import write_pty_payload
 
 type _PromptResult = tuple[Literal["ok"], str] | tuple[Literal["err"], str]
@@ -59,7 +59,7 @@ async def ask_into_pty(
     label = message.strip() or ("Secret" if secret else "Input")
     try:
         # Validate session before blocking the human.
-        _ = PtyManager.refresh(session_id)
+        _ = active_pty_manager().refresh(session_id)
     except WorkspaceError as exc:
         return f"error: {exc}"
 

src/plyngent/tools/process/close_pty.py

@@ -4,7 +4,7 @@ import asyncio
 
 from plyngent.agent import ToolTag, tool
 
-from .pty_session import PtyManager, format_close_result
+from .pty_session import active_pty_manager, format_close_result
 
 
 @tool(tags=ToolTag.LOCAL | ToolTag.INSTANCE_STATE)
@@ -13,5 +13,5 @@ async def close_pty(session_id: int) -> str:
 
     Runs off the event loop so grace sleeps do not freeze the chat UI.
     """
-    result = await asyncio.to_thread(PtyManager.close, session_id)
+    result = await asyncio.to_thread(active_pty_manager().close, session_id)
     return format_close_result(result)

src/plyngent/tools/process/open_pty.py

@@ -5,7 +5,7 @@ import shlex
 from plyngent.agent import ToolTag, tool
 from plyngent.tools.workspace import WorkspaceError
 
-from .pty_session import PtyManager
+from .pty_session import active_pty_manager
 
 
 @tool(tags=ToolTag.LOCAL | ToolTag.INSTANCE_STATE | ToolTag.YOLO)
@@ -17,7 +17,7 @@ async def open_pty(command: list[str], *, cwd: str = ".") -> str:
     data (marker) and a non-zero exit_code.
     """
     try:
-        session = PtyManager.open(command, cwd=cwd)
+        session = active_pty_manager().open(command, cwd=cwd)
     except WorkspaceError as exc:
         return f"error: {exc}"
     except OSError as exc:

src/plyngent/tools/process/pty_session.py

@@ -5,7 +5,7 @@ from __future__ import annotations
 import time
 from dataclasses import dataclass, field
 from threading import Lock
-from typing import TYPE_CHECKING, ClassVar
+from typing import TYPE_CHECKING, ClassVar, cast
 
 from plyngent.tools.process.pty_backend import PtyHandle, pty_available, spawn_pty
 from plyngent.tools.process.pty_terminal import sanitize_pty_output_for_tool
@@ -393,6 +393,20 @@ class PtyManager:
             _ = cls.close(session_id)
 
 
+def active_pty_manager() -> type[PtyManager]:
+    """Return the PTY manager for the bound instance, else the process default.
+
+    Today :class:`PtyManager` is process-global; the instance facet is a stable
+    host API so tools do not hard-code the class forever.
+    """
+    from plyngent.tools.context import get_instance
+
+    instance = get_instance()
+    if instance is not None:
+        return cast("type[PtyManager]", instance.pty)
+    return PtyManager
+
+
 def format_read_result(result: PtyReadResult) -> str:
     exit_disp = "" if result.exit_code is None else str(result.exit_code)
     lines = [

src/plyngent/tools/process/read_pty.py

@@ -5,7 +5,7 @@ import asyncio
 from plyngent.agent import ToolTag, tool
 from plyngent.tools.workspace import WorkspaceError
 
-from .pty_session import DEFAULT_PTY_READ_BYTES, PtyManager, format_read_result
+from .pty_session import DEFAULT_PTY_READ_BYTES, active_pty_manager, format_read_result
 
 # Cap per-call wait so a misbehaving tool arg cannot stall forever even off-loop.
 _MAX_READ_TIMEOUT = 120.0
@@ -34,7 +34,7 @@ async def read_pty(
     wait = min(max(0.0, timeout), _MAX_READ_TIMEOUT)
     try:
         result = await asyncio.to_thread(
-            PtyManager.read,
+            active_pty_manager().read,
             session_id,
             max_bytes=max_bytes,
             timeout=wait,

src/plyngent/tools/process/write_pty.py

@@ -3,13 +3,14 @@ from __future__ import annotations
 from plyngent.agent import ToolTag, tool
 from plyngent.tools.workspace import WorkspaceError
 
-from .pty_session import PtyManager
+from .pty_session import active_pty_manager
 
 
 def write_pty_payload(session_id: int, raw: str) -> str:
     """Write raw bytes (as str) to the PTY and format the tool status string."""
-    PtyManager.write(session_id, raw)
+    manager = active_pty_manager()
-    session = PtyManager.refresh(session_id)
+    manager.write(session_id, raw)
+    session = manager.refresh(session_id)
     exit_disp = "" if session.exit_code is None else str(session.exit_code)
     return "\n".join(
         [