core/tools: prefer instance workspace and session todo views

This commit is contained in:
2026-07-24 15:02:52 +08:00
parent 12c8adda3a
commit 7253ca84dd
4 changed files with 101 additions and 8 deletions
+12 -3
View File
@@ -7,12 +7,21 @@ from contextvars import ContextVar, Token
from dataclasses import dataclass, field
from typing import TYPE_CHECKING, Any
from plyngent.tools.view import PersistentDataView, session_data_view
if TYPE_CHECKING:
from collections.abc import Generator
from pathlib import Path
from plyngent.agent.todo_stack import TodoStack
from plyngent.tools.view import PersistentDataView
def _default_instance_data() -> PersistentDataView[Any]:
return session_data_view()
def _default_session_data() -> PersistentDataView[Any]:
return session_data_view()
@dataclass
@@ -22,7 +31,7 @@ class InstanceState:
# Optional fixed facets (workspace path still also lives in workspace module
# during migration; hosts should set both until globals retire).
workspace_root: Path | None = None
data: PersistentDataView[Any] | None = None
data: PersistentDataView[Any] = field(default_factory=_default_instance_data)
# Ephemeral process maps (PTY etc.) may hang here later.
extras: dict[str, Any] = field(default_factory=dict)
@@ -40,7 +49,7 @@ class SessionState:
"""One chat session for SESSION_STATE tools."""
session_id: int | str | None = None
data: PersistentDataView[Any] | None = None
data: PersistentDataView[Any] = field(default_factory=_default_session_data)
# Live domain object during migration (also under data["todo"] when views bind).
todo: TodoStack | None = None
# Soft-confirm grants: tool_name → True (Phase 1 key is tool name in session).
+17 -3
View File
@@ -27,12 +27,26 @@ def get_todo_stack() -> TodoStack | None:
def _require_stack() -> TodoStack:
"""Prefer session-bound todo; fall back to process bind during migration."""
"""Prefer session-bound todo; fall back to process bind during migration.
Order: ``session.todo`` facet → ``session.data["todo"].typed(TodoStack)`` when a
view is open/bound → process ``set_todo_stack`` global.
"""
from plyngent.agent.todo_stack import TodoStack
from plyngent.tools.context import get_session
session = get_session()
if session is not None and session.todo is not None:
return session.todo
if session is not None:
if session.todo is not None:
return session.todo
try:
# Prefer live domain object when a txn is open; otherwise skip.
todo = session.data["todo"].typed(TodoStack)
except RuntimeError:
todo = None
if isinstance(todo, TodoStack):
session.todo = todo
return todo
if _stack is None:
msg = "todo stack is not available in this session"
raise RuntimeError(msg)
+25 -2
View File
@@ -69,17 +69,35 @@ _state = _WorkspaceState()
def set_workspace_root(root: Path | str) -> Path:
"""Set the workspace root used by tools; returns the resolved root."""
"""Set the workspace root used by tools; returns the resolved root.
Writes the process-global root and, when an :class:`~plyngent.tools.context.InstanceState`
is bound, mirrors onto ``instance.workspace_root``.
"""
path = Path(root).expanduser().resolve()
if not path.is_dir():
msg = f"workspace root is not a directory: {path}"
raise WorkspaceError(msg)
_state.root = path
from plyngent.tools.context import get_instance
instance = get_instance()
if instance is not None:
instance.workspace_root = path
return path
def get_workspace_root() -> Path:
"""Return the configured workspace root."""
"""Return the configured workspace root.
Prefers a bound instance's ``workspace_root`` when set; otherwise the
process-global root from :func:`set_workspace_root`.
"""
from plyngent.tools.context import get_instance
instance = get_instance()
if instance is not None and instance.workspace_root is not None:
return instance.workspace_root
if _state.root is None:
msg = "workspace root is not set; call set_workspace_root() first"
raise WorkspaceError(msg)
@@ -89,6 +107,11 @@ def get_workspace_root() -> Path:
def clear_workspace_root() -> None:
"""Clear workspace root (mainly for tests). Does not clear allowlist."""
_state.root = None
from plyngent.tools.context import get_instance
instance = get_instance()
if instance is not None:
instance.workspace_root = None
def set_path_denylist(patterns: list[str] | tuple[str, ...] | None) -> None:
@@ -0,0 +1,47 @@
"""Workspace root prefers bound InstanceState when set."""
from __future__ import annotations
from pathlib import Path
import pytest
from plyngent.tools.context import InstanceState, bind_instance
from plyngent.tools.workspace import (
WorkspaceError,
clear_workspace_root,
get_workspace_root,
set_workspace_root,
)
def test_get_workspace_prefers_instance(tmp_path: Path) -> None:
clear_workspace_root()
other = tmp_path / "other"
other.mkdir()
_ = set_workspace_root(tmp_path)
instance = InstanceState(workspace_root=other)
with bind_instance(instance):
assert get_workspace_root() == other.resolve()
assert get_workspace_root() == tmp_path.resolve()
clear_workspace_root()
def test_set_workspace_mirrors_to_bound_instance(tmp_path: Path) -> None:
clear_workspace_root()
instance = InstanceState()
with bind_instance(instance):
path = set_workspace_root(tmp_path)
assert instance.workspace_root == path
assert get_workspace_root() == path
clear_workspace_root()
def test_clear_clears_bound_instance(tmp_path: Path) -> None:
instance = InstanceState()
with bind_instance(instance):
_ = set_workspace_root(tmp_path)
clear_workspace_root()
assert instance.workspace_root is None
with pytest.raises(WorkspaceError):
_ = get_workspace_root()