mirror of
https://github.com/agent0ai/agent-zero.git
synced 2026-07-09 17:08:29 +00:00
Time Travel workspace selector + LiteLLM globals
Add selectable workspace support to the Time Travel plugin and introduce normalized global LiteLLM configuration handling. models.py: add DEFAULT_LITELLM_GLOBAL_KWARGS and helpers to normalize, load, apply, and merge LiteLLM global kwargs; call configure_litellm/set per-call merges in LiteLLM wrappers so framework defaults, configured globals, and per-call overrides combine correctly. Add tests to assert merging and runtime application. plugins/_time_travel: update docs and plugin metadata to describe workdir/project selection. API handlers now accept workspace_id and a new history_workspaces endpoint lists selectable workspaces. helpers/time_travel: implement workspace listing, selection, and resolution logic (workdir/project options, availability/locking, default selection). UI: add workspace picker to time-travel panel, wire selection/load into time-travel store, pass workspace_id to API calls, and add click hook to open the panel. Update styles and refresh/load behavior accordingly. Tests: extend tests for LiteLLM global kwargs merging and adapt stream test to expect merged params; add tests for selectable workspaces, default selection, and locked external workdir handling.
This commit is contained in:
frdel
parent
60462eabbd
commit
77f7aa0274
17 changed files with 580 additions and 79 deletions
|
|
@ -101,6 +101,7 @@ Key Files:
|
|||
- helpers/plugins.py: Plugin discovery and configuration logic.
|
||||
- webui/js/AlpineStore.js: Store factory for reactive frontend state.
|
||||
- helpers/api.py: Base class for all API endpoints.
|
||||
- models.py: LLM provider configuration and LiteLLM wrappers; framework LiteLLM defaults such as `drop_params=True` are merged with `litellm_global_kwargs`, configured values override framework defaults, and the merged kwargs are applied at LiteLLM module and per-call boundaries.
|
||||
- scripts/openrouter_release_notes_system_prompt.md: Editable system prompt used to generate GitHub release notes during Docker publishing.
|
||||
- knowledge/main/about/: Agent self-knowledge files, indexed into the vector DB for runtime recall. Not user-facing docs - written for the agent's internal reference.
|
||||
- webui/components/AGENTS.md: DOX contract for Alpine component architecture.
|
||||
|
|
|
|||
129
models.py
129
models.py
|
|
@ -45,6 +45,48 @@ from sentence_transformers import SentenceTransformer
|
|||
from pydantic import ConfigDict
|
||||
|
||||
|
||||
DEFAULT_LITELLM_GLOBAL_KWARGS: dict[str, Any] = {
|
||||
"drop_params": True,
|
||||
}
|
||||
|
||||
|
||||
def _normalize_litellm_kwargs(values: dict[str, Any]) -> dict[str, Any]:
|
||||
# Normalize .env/UI-style scalar strings into native types for LiteLLM.
|
||||
result: dict[str, Any] = {}
|
||||
for k, v in values.items():
|
||||
if isinstance(v, str):
|
||||
stripped = v.strip()
|
||||
lowered = stripped.lower()
|
||||
if lowered == "true":
|
||||
result[k] = True
|
||||
elif lowered == "false":
|
||||
result[k] = False
|
||||
elif lowered in ("none", "null"):
|
||||
result[k] = None
|
||||
else:
|
||||
try:
|
||||
result[k] = int(stripped)
|
||||
except ValueError:
|
||||
try:
|
||||
result[k] = float(stripped)
|
||||
except ValueError:
|
||||
result[k] = v
|
||||
else:
|
||||
result[k] = v
|
||||
return result
|
||||
|
||||
|
||||
def get_litellm_global_kwargs() -> dict[str, Any]:
|
||||
kwargs = _normalize_litellm_kwargs(DEFAULT_LITELLM_GLOBAL_KWARGS)
|
||||
try:
|
||||
configured = settings.get_settings().get("litellm_global_kwargs", {}) # type: ignore[union-attr]
|
||||
except Exception:
|
||||
configured = {}
|
||||
if isinstance(configured, dict):
|
||||
kwargs.update(_normalize_litellm_kwargs(configured))
|
||||
return kwargs
|
||||
|
||||
|
||||
# keep provider logging quiet in normal operation
|
||||
def turn_off_logging():
|
||||
os.environ["LITELLM_LOG"] = "ERROR" # only errors
|
||||
|
|
@ -55,9 +97,30 @@ def turn_off_logging():
|
|||
logging.getLogger(name).setLevel(logging.ERROR)
|
||||
|
||||
|
||||
def set_litellm_params():
|
||||
global_kwargs = get_litellm_global_kwargs()
|
||||
for key, value in global_kwargs.items():
|
||||
setattr(litellm, key, value)
|
||||
return global_kwargs
|
||||
|
||||
|
||||
def configure_litellm():
|
||||
turn_off_logging()
|
||||
set_litellm_params()
|
||||
|
||||
|
||||
def _merge_litellm_call_kwargs(*overrides: dict[str, Any] | None) -> dict[str, Any]:
|
||||
kwargs = get_litellm_global_kwargs()
|
||||
for override in overrides:
|
||||
if isinstance(override, dict):
|
||||
kwargs.update(override)
|
||||
return kwargs
|
||||
|
||||
|
||||
# init
|
||||
load_dotenv()
|
||||
turn_off_logging()
|
||||
configure_litellm()
|
||||
|
||||
|
||||
class ModelType(Enum):
|
||||
CHAT = "Chat"
|
||||
|
|
@ -390,13 +453,16 @@ class LiteLLMChatWrapper(SimpleChatModel):
|
|||
) -> str:
|
||||
import asyncio
|
||||
|
||||
configure_litellm()
|
||||
msgs = self._convert_messages(messages)
|
||||
|
||||
# Apply rate limiting if configured
|
||||
apply_rate_limiter_sync(self.a0_model_conf, str(msgs))
|
||||
|
||||
# Call the model
|
||||
call_kwargs = _without_stream_kwarg({**self.kwargs, **kwargs})
|
||||
call_kwargs = _without_stream_kwarg(
|
||||
_merge_litellm_call_kwargs(self.kwargs, kwargs)
|
||||
)
|
||||
resp = completion(
|
||||
model=self.model_name, messages=msgs, stop=stop, **call_kwargs
|
||||
)
|
||||
|
|
@ -415,13 +481,16 @@ class LiteLLMChatWrapper(SimpleChatModel):
|
|||
) -> Iterator[ChatGenerationChunk]:
|
||||
import asyncio
|
||||
|
||||
configure_litellm()
|
||||
msgs = self._convert_messages(messages)
|
||||
|
||||
# Apply rate limiting if configured
|
||||
apply_rate_limiter_sync(self.a0_model_conf, str(msgs))
|
||||
|
||||
result = ChatGenerationResult()
|
||||
call_kwargs = _without_stream_kwarg({**self.kwargs, **kwargs})
|
||||
call_kwargs = _without_stream_kwarg(
|
||||
_merge_litellm_call_kwargs(self.kwargs, kwargs)
|
||||
)
|
||||
|
||||
for chunk in completion(
|
||||
model=self.model_name,
|
||||
|
|
@ -447,13 +516,16 @@ class LiteLLMChatWrapper(SimpleChatModel):
|
|||
run_manager: Optional[AsyncCallbackManagerForLLMRun] = None,
|
||||
**kwargs: Any,
|
||||
) -> AsyncIterator[ChatGenerationChunk]:
|
||||
configure_litellm()
|
||||
msgs = self._convert_messages(messages)
|
||||
|
||||
# Apply rate limiting if configured
|
||||
await apply_rate_limiter(self.a0_model_conf, str(msgs))
|
||||
|
||||
result = ChatGenerationResult()
|
||||
call_kwargs = _without_stream_kwarg({**self.kwargs, **kwargs})
|
||||
call_kwargs = _without_stream_kwarg(
|
||||
_merge_litellm_call_kwargs(self.kwargs, kwargs)
|
||||
)
|
||||
|
||||
response = await acompletion(
|
||||
model=self.model_name,
|
||||
|
|
@ -488,7 +560,7 @@ class LiteLLMChatWrapper(SimpleChatModel):
|
|||
**kwargs: Any,
|
||||
) -> Tuple[str, str]:
|
||||
|
||||
turn_off_logging()
|
||||
configure_litellm()
|
||||
|
||||
if not messages:
|
||||
messages = []
|
||||
|
|
@ -507,7 +579,9 @@ class LiteLLMChatWrapper(SimpleChatModel):
|
|||
)
|
||||
|
||||
# Prepare call kwargs and retry config (strip A0-only params before calling LiteLLM)
|
||||
call_kwargs: dict[str, Any] = _without_stream_kwarg({**self.kwargs, **kwargs})
|
||||
call_kwargs: dict[str, Any] = _without_stream_kwarg(
|
||||
_merge_litellm_call_kwargs(self.kwargs, kwargs)
|
||||
)
|
||||
max_retries: int = int(call_kwargs.pop("a0_retry_attempts", 2))
|
||||
retry_delay_s: float = float(call_kwargs.pop("a0_retry_delay_seconds", 1.5))
|
||||
stream = reasoning_callback is not None or response_callback is not None or tokens_callback is not None
|
||||
|
|
@ -610,20 +684,30 @@ class LiteLLMEmbeddingWrapper(Embeddings):
|
|||
self.a0_model_conf = model_config
|
||||
|
||||
def embed_documents(self, texts: List[str]) -> List[List[float]]:
|
||||
configure_litellm()
|
||||
# Apply rate limiting if configured
|
||||
apply_rate_limiter_sync(self.a0_model_conf, " ".join(texts))
|
||||
|
||||
resp = embedding(model=self.model_name, input=texts, **self.kwargs)
|
||||
resp = embedding(
|
||||
model=self.model_name,
|
||||
input=texts,
|
||||
**_merge_litellm_call_kwargs(self.kwargs),
|
||||
)
|
||||
return [
|
||||
item.get("embedding") if isinstance(item, dict) else item.embedding # type: ignore
|
||||
for item in resp.data # type: ignore
|
||||
]
|
||||
|
||||
def embed_query(self, text: str) -> List[float]:
|
||||
configure_litellm()
|
||||
# Apply rate limiting if configured
|
||||
apply_rate_limiter_sync(self.a0_model_conf, text)
|
||||
|
||||
resp = embedding(model=self.model_name, input=[text], **self.kwargs)
|
||||
resp = embedding(
|
||||
model=self.model_name,
|
||||
input=[text],
|
||||
**_merge_litellm_call_kwargs(self.kwargs),
|
||||
)
|
||||
item = resp.data[0] # type: ignore
|
||||
return item.get("embedding") if isinstance(item, dict) else item.embedding # type: ignore
|
||||
|
||||
|
|
@ -781,22 +865,6 @@ def _adjust_call_args(provider_name: str, model_name: str, kwargs: dict):
|
|||
def _merge_provider_defaults(
|
||||
provider_type: ProviderModelType, original_provider: str, kwargs: dict
|
||||
) -> tuple[str, dict]:
|
||||
# Normalize .env-style numeric strings (e.g., "timeout=30") into ints/floats for LiteLLM
|
||||
def _normalize_values(values: dict) -> dict:
|
||||
result: dict[str, Any] = {}
|
||||
for k, v in values.items():
|
||||
if isinstance(v, str):
|
||||
try:
|
||||
result[k] = int(v)
|
||||
except ValueError:
|
||||
try:
|
||||
result[k] = float(v)
|
||||
except ValueError:
|
||||
result[k] = v
|
||||
else:
|
||||
result[k] = v
|
||||
return result
|
||||
|
||||
provider_name = original_provider # default: unchanged
|
||||
cfg = get_provider_config(provider_type, original_provider)
|
||||
if cfg:
|
||||
|
|
@ -814,14 +882,11 @@ def _merge_provider_defaults(
|
|||
if key and key not in ("None", "NA"):
|
||||
kwargs["api_key"] = key
|
||||
|
||||
# Merge LiteLLM global kwargs (timeouts, stream_timeout, etc.)
|
||||
try:
|
||||
global_kwargs = settings.get_settings().get("litellm_global_kwargs", {}) # type: ignore[union-attr]
|
||||
except Exception:
|
||||
global_kwargs = {}
|
||||
if isinstance(global_kwargs, dict):
|
||||
for k, v in _normalize_values(global_kwargs).items():
|
||||
kwargs.setdefault(k, v)
|
||||
# Merge LiteLLM global kwargs. Framework defaults are merged first, then
|
||||
# configured global kwargs override those defaults; explicit provider/model
|
||||
# kwargs still keep priority via setdefault.
|
||||
for k, v in get_litellm_global_kwargs().items():
|
||||
kwargs.setdefault(k, v)
|
||||
|
||||
return provider_name, kwargs
|
||||
|
||||
|
|
|
|||
|
|
@ -2,18 +2,18 @@
|
|||
|
||||
## Purpose
|
||||
|
||||
- Own Agent Zero workspace history, diff inspection, travel, snapshots, and revert for active `/a0/usr` workspaces.
|
||||
- Own Agent Zero workspace history, diff inspection, travel, snapshots, and revert for selectable Agent Zero workdir/project workspaces under `/a0/usr`.
|
||||
|
||||
## Ownership
|
||||
|
||||
- `helpers/time_travel.py` owns history storage, diff, travel, snapshot, preview, and revert mechanics.
|
||||
- `api/` owns history list, diff, preview, revert, snapshot, and travel endpoints.
|
||||
- `webui/` owns the time-travel panel, store, main surface, and thumbnail.
|
||||
- `api/` owns selectable workspace list, history list, diff, preview, revert, snapshot, and travel endpoints.
|
||||
- `webui/` owns the time-travel panel, workspace selector, store, main surface, and thumbnail.
|
||||
- `plugin.yaml` and `extensions/` own metadata and hook contributions.
|
||||
|
||||
## Local Contracts
|
||||
|
||||
- Keep history operations scoped to Agent Zero-owned workspaces.
|
||||
- Keep history operations scoped to Agent Zero-owned workdir/project workspaces.
|
||||
- Revert and travel operations must avoid unintended writes outside managed workspace paths.
|
||||
- Preserve enough metadata for clear preview and diff inspection before destructive actions.
|
||||
|
||||
|
|
@ -23,7 +23,7 @@
|
|||
|
||||
## Verification
|
||||
|
||||
- Smoke-test snapshot, list, diff, preview, travel, and revert flows after changes.
|
||||
- Smoke-test workspace selection, snapshot, list, diff, preview, travel, and revert flows after changes.
|
||||
|
||||
## Child DOX Index
|
||||
|
||||
|
|
|
|||
|
|
@ -12,8 +12,13 @@ from plugins._time_travel.helpers.time_travel import (
|
|||
class HistoryDiff(ApiHandler):
|
||||
async def process(self, input: dict, request: Request) -> dict | Response:
|
||||
context_id = str(input.get("context_id") or "").strip()
|
||||
workspace_id = str(input.get("workspace_id") or "").strip()
|
||||
try:
|
||||
workspace = resolve_workspace(context_id, context_loader=self.use_context)
|
||||
workspace = resolve_workspace(
|
||||
context_id,
|
||||
workspace_id=workspace_id,
|
||||
context_loader=self.use_context,
|
||||
)
|
||||
return TimeTravelService(workspace).history_diff(
|
||||
commit_hash=str(input.get("commit_hash") or ""),
|
||||
path=str(input.get("path") or ""),
|
||||
|
|
|
|||
|
|
@ -13,8 +13,13 @@ from plugins._time_travel.helpers.time_travel import (
|
|||
class HistoryList(ApiHandler):
|
||||
async def process(self, input: dict, request: Request) -> dict | Response:
|
||||
context_id = str(input.get("context_id") or "").strip()
|
||||
workspace_id = str(input.get("workspace_id") or "").strip()
|
||||
try:
|
||||
workspace = resolve_workspace(context_id, context_loader=self.use_context)
|
||||
workspace = resolve_workspace(
|
||||
context_id,
|
||||
workspace_id=workspace_id,
|
||||
context_loader=self.use_context,
|
||||
)
|
||||
return TimeTravelService(workspace).history_list(
|
||||
limit=int(input.get("limit") or 100),
|
||||
offset=int(input.get("offset") or 0),
|
||||
|
|
|
|||
|
|
@ -12,8 +12,13 @@ from plugins._time_travel.helpers.time_travel import (
|
|||
class HistoryPreview(ApiHandler):
|
||||
async def process(self, input: dict, request: Request) -> dict | Response:
|
||||
context_id = str(input.get("context_id") or "").strip()
|
||||
workspace_id = str(input.get("workspace_id") or "").strip()
|
||||
try:
|
||||
workspace = resolve_workspace(context_id, context_loader=self.use_context)
|
||||
workspace = resolve_workspace(
|
||||
context_id,
|
||||
workspace_id=workspace_id,
|
||||
context_loader=self.use_context,
|
||||
)
|
||||
return TimeTravelService(workspace).preview(
|
||||
operation=str(input.get("operation") or ""),
|
||||
commit_hash=str(input.get("commit_hash") or ""),
|
||||
|
|
|
|||
|
|
@ -12,8 +12,13 @@ from plugins._time_travel.helpers.time_travel import (
|
|||
class HistoryRevert(ApiHandler):
|
||||
async def process(self, input: dict, request: Request) -> dict | Response:
|
||||
context_id = str(input.get("context_id") or "").strip()
|
||||
workspace_id = str(input.get("workspace_id") or "").strip()
|
||||
try:
|
||||
workspace = resolve_workspace(context_id, context_loader=self.use_context)
|
||||
workspace = resolve_workspace(
|
||||
context_id,
|
||||
workspace_id=workspace_id,
|
||||
context_loader=self.use_context,
|
||||
)
|
||||
return TimeTravelService(workspace).revert(
|
||||
commit_hash=str(input.get("commit_hash") or ""),
|
||||
metadata=input.get("metadata") if isinstance(input.get("metadata"), dict) else {},
|
||||
|
|
|
|||
|
|
@ -13,8 +13,13 @@ from plugins._time_travel.helpers.time_travel import (
|
|||
class HistorySnapshot(ApiHandler):
|
||||
async def process(self, input: dict, request: Request) -> dict | Response:
|
||||
context_id = str(input.get("context_id") or "").strip()
|
||||
workspace_id = str(input.get("workspace_id") or "").strip()
|
||||
try:
|
||||
workspace = resolve_workspace(context_id, context_loader=self.use_context)
|
||||
workspace = resolve_workspace(
|
||||
context_id,
|
||||
workspace_id=workspace_id,
|
||||
context_loader=self.use_context,
|
||||
)
|
||||
snapshot = TimeTravelService(workspace).snapshot(
|
||||
trigger=str(input.get("trigger") or "manual"),
|
||||
message=str(input.get("message") or ""),
|
||||
|
|
|
|||
|
|
@ -12,8 +12,13 @@ from plugins._time_travel.helpers.time_travel import (
|
|||
class HistoryTravel(ApiHandler):
|
||||
async def process(self, input: dict, request: Request) -> dict | Response:
|
||||
context_id = str(input.get("context_id") or "").strip()
|
||||
workspace_id = str(input.get("workspace_id") or "").strip()
|
||||
try:
|
||||
workspace = resolve_workspace(context_id, context_loader=self.use_context)
|
||||
workspace = resolve_workspace(
|
||||
context_id,
|
||||
workspace_id=workspace_id,
|
||||
context_loader=self.use_context,
|
||||
)
|
||||
return TimeTravelService(workspace).travel(
|
||||
commit_hash=str(input.get("commit_hash") or ""),
|
||||
metadata=input.get("metadata") if isinstance(input.get("metadata"), dict) else {},
|
||||
|
|
|
|||
11
plugins/_time_travel/api/history_workspaces.py
Normal file
11
plugins/_time_travel/api/history_workspaces.py
Normal file
|
|
@ -0,0 +1,11 @@
|
|||
from __future__ import annotations
|
||||
|
||||
from helpers.api import ApiHandler, Request, Response
|
||||
from plugins._time_travel.helpers.time_travel import list_selectable_workspaces
|
||||
|
||||
|
||||
class HistoryWorkspaces(ApiHandler):
|
||||
async def process(self, input: dict, request: Request) -> dict | Response:
|
||||
context_id = str(input.get("context_id") or "").strip()
|
||||
data = list_selectable_workspaces(context_id, context_loader=self.use_context)
|
||||
return {"ok": True, **data}
|
||||
|
|
@ -5,7 +5,7 @@
|
|||
class="dropdown-item"
|
||||
id="time-travel-dropdown"
|
||||
title="Time Travel"
|
||||
@click="ensureModalOpen('/plugins/_time_travel/webui/main.html'); $store.sidebar.menuClose()">
|
||||
@click="$store.timeTravel?.onOpen?.(); ensureModalOpen('/plugins/_time_travel/webui/main.html'); $store.sidebar.menuClose()">
|
||||
<span class="material-symbols-outlined">history</span>
|
||||
<span>Time Travel</span>
|
||||
</button>
|
||||
|
|
|
|||
|
|
@ -210,7 +210,132 @@ def canonical_workspace_display_path(display_path: str) -> str:
|
|||
return (canonical if canonical.startswith("/a0") else normalized).rstrip("/") or canonical
|
||||
|
||||
|
||||
def resolve_workspace(context_id: str = "", *, context_loader=None) -> WorkspaceInfo:
|
||||
def configured_workdir_display_path() -> str:
|
||||
from helpers import settings
|
||||
|
||||
configured = str(settings.get_settings().get("workdir_path") or "")
|
||||
fallback = files.normalize_a0_path(files.get_abs_path("usr/workdir"))
|
||||
return canonical_workspace_display_path(configured or fallback)
|
||||
|
||||
|
||||
def _workspace_option(
|
||||
*,
|
||||
kind: str,
|
||||
display_path: str,
|
||||
label: str,
|
||||
name: str = "",
|
||||
title: str = "",
|
||||
project_name: str = "",
|
||||
color: str = "",
|
||||
) -> dict[str, Any]:
|
||||
normalized = canonical_workspace_display_path(display_path)
|
||||
available = is_inside_usr_display(normalized)
|
||||
error = (
|
||||
"" if available else "Time Travel is only available for workspaces inside /a0/usr."
|
||||
)
|
||||
return {
|
||||
"id": workspace_id_for(normalized),
|
||||
"kind": kind,
|
||||
"name": name,
|
||||
"title": title or label,
|
||||
"label": label,
|
||||
"display_path": normalized.rstrip("/") or normalized,
|
||||
"path": normalized.rstrip("/") or normalized,
|
||||
"project_name": project_name,
|
||||
"color": color,
|
||||
"available": available,
|
||||
"locked": not available,
|
||||
"error": error,
|
||||
}
|
||||
|
||||
|
||||
def list_selectable_workspaces(
|
||||
context_id: str = "",
|
||||
*,
|
||||
context_loader=None,
|
||||
) -> dict[str, Any]:
|
||||
from helpers import projects
|
||||
|
||||
context_id = str(context_id or "").strip()
|
||||
workspaces: list[dict[str, Any]] = []
|
||||
seen_ids: set[str] = set()
|
||||
|
||||
def add_workspace(option: dict[str, Any]) -> None:
|
||||
option_id = str(option.get("id") or "")
|
||||
if not option_id or option_id in seen_ids:
|
||||
return
|
||||
seen_ids.add(option_id)
|
||||
workspaces.append(option)
|
||||
|
||||
add_workspace(
|
||||
_workspace_option(
|
||||
kind="workdir",
|
||||
display_path=configured_workdir_display_path(),
|
||||
label="User working directory",
|
||||
name="workdir",
|
||||
title="User working directory",
|
||||
)
|
||||
)
|
||||
|
||||
for project in projects.get_active_projects_list() or []:
|
||||
project_name = str(project.get("name") or "").strip()
|
||||
if not project_name:
|
||||
continue
|
||||
title = str(project.get("title") or project_name)
|
||||
add_workspace(
|
||||
_workspace_option(
|
||||
kind="project",
|
||||
display_path=files.normalize_a0_path(
|
||||
projects.get_project_folder(project_name)
|
||||
),
|
||||
label=title,
|
||||
name=project_name,
|
||||
title=title,
|
||||
project_name=project_name,
|
||||
color=str(project.get("color") or ""),
|
||||
)
|
||||
)
|
||||
|
||||
default_workspace_id = ""
|
||||
try:
|
||||
default_workspace_id = _resolve_context_workspace(
|
||||
context_id,
|
||||
context_loader=context_loader,
|
||||
).id
|
||||
except TimeTravelError:
|
||||
default_workspace_id = ""
|
||||
|
||||
if default_workspace_id not in seen_ids:
|
||||
default_workspace_id = ""
|
||||
if not default_workspace_id and workspaces:
|
||||
default_workspace_id = str(workspaces[0].get("id") or "")
|
||||
|
||||
return {
|
||||
"context_id": context_id,
|
||||
"workspaces": workspaces,
|
||||
"default_workspace_id": default_workspace_id,
|
||||
}
|
||||
|
||||
|
||||
def _selectable_workspace_by_id(
|
||||
workspace_id: str,
|
||||
context_id: str = "",
|
||||
*,
|
||||
context_loader=None,
|
||||
) -> dict[str, Any] | None:
|
||||
wanted = str(workspace_id or "").strip()
|
||||
if not wanted:
|
||||
return None
|
||||
for workspace in list_selectable_workspaces(
|
||||
context_id,
|
||||
context_loader=context_loader,
|
||||
)["workspaces"]:
|
||||
if str(workspace.get("id") or "") == wanted:
|
||||
return workspace
|
||||
return None
|
||||
|
||||
|
||||
def _resolve_context_workspace(context_id: str = "", *, context_loader=None) -> WorkspaceInfo:
|
||||
from helpers import projects, settings
|
||||
|
||||
context_id = str(context_id or "").strip()
|
||||
|
|
@ -246,9 +371,36 @@ def resolve_workspace(context_id: str = "", *, context_loader=None) -> Workspace
|
|||
)
|
||||
|
||||
|
||||
def resolve_workspace_for_path_hint(path_hint: str) -> WorkspaceInfo | None:
|
||||
from helpers import settings
|
||||
def resolve_workspace(
|
||||
context_id: str = "",
|
||||
*,
|
||||
workspace_id: str = "",
|
||||
context_loader=None,
|
||||
) -> WorkspaceInfo:
|
||||
workspace_id = str(workspace_id or "").strip()
|
||||
if not workspace_id:
|
||||
return _resolve_context_workspace(context_id, context_loader=context_loader)
|
||||
|
||||
selected = _selectable_workspace_by_id(
|
||||
workspace_id,
|
||||
context_id,
|
||||
context_loader=context_loader,
|
||||
)
|
||||
if not selected:
|
||||
raise WorkspaceRejectedError("Selected Time Travel workspace is not available.")
|
||||
if selected.get("locked") or selected.get("available") is False:
|
||||
raise WorkspaceRejectedError(
|
||||
str(selected.get("error") or "Selected Time Travel workspace is not available.")
|
||||
)
|
||||
|
||||
return _workspace_from_display(
|
||||
str(selected.get("display_path") or selected.get("path") or ""),
|
||||
project_name=str(selected.get("project_name") or ""),
|
||||
context_id=str(context_id or "").strip(),
|
||||
)
|
||||
|
||||
|
||||
def resolve_workspace_for_path_hint(path_hint: str) -> WorkspaceInfo | None:
|
||||
normalized = canonical_workspace_display_path(path_hint)
|
||||
if not is_inside_usr_display(normalized):
|
||||
return None
|
||||
|
|
@ -258,8 +410,7 @@ def resolve_workspace_for_path_hint(path_hint: str) -> WorkspaceInfo | None:
|
|||
project_display = f"/a0/usr/projects/{parts[3]}"
|
||||
return _workspace_from_display(project_display, project_name=parts[3])
|
||||
|
||||
configured = str(settings.get_settings().get("workdir_path") or "")
|
||||
workdir_display = canonical_workspace_display_path(configured or files.normalize_a0_path(files.get_abs_path("usr/workdir")))
|
||||
workdir_display = configured_workdir_display_path()
|
||||
if normalized == workdir_display or normalized.startswith(workdir_display.rstrip("/") + "/"):
|
||||
return _workspace_from_display(workdir_display)
|
||||
|
||||
|
|
|
|||
|
|
@ -1,6 +1,6 @@
|
|||
name: _time_travel
|
||||
title: Time Travel
|
||||
description: Agent Zero-owned workspace history, diff inspection, travel, and revert for active /a0/usr workspaces.
|
||||
description: Agent Zero-owned workdir/project history, diff inspection, travel, and revert for /a0/usr workspaces.
|
||||
version: 0.1.0
|
||||
always_enabled: false
|
||||
settings_sections: []
|
||||
|
|
|
|||
|
|
@ -9,16 +9,29 @@
|
|||
<template x-if="$store.timeTravel">
|
||||
<div class="time-travel-shell">
|
||||
<div class="time-travel-toolbar">
|
||||
<div class="time-travel-title">
|
||||
<span class="material-symbols-outlined">history</span>
|
||||
<span>Time Travel</span>
|
||||
<div class="time-travel-workspace-picker" :title="$store.timeTravel.workspacePath || $store.timeTravel.workspaceOptionLabel($store.timeTravel.selectedWorkspace())">
|
||||
<span class="material-symbols-outlined" aria-hidden="true">folder_open</span>
|
||||
<select
|
||||
aria-label="Workspace"
|
||||
:value="$store.timeTravel.selectedWorkspaceId"
|
||||
@change="$store.timeTravel.selectWorkspace($event.target.value)"
|
||||
:disabled="$store.timeTravel.workspaceLoading || $store.timeTravel.loading || $store.timeTravel.busy"
|
||||
>
|
||||
<option value="" x-show="$store.timeTravel.workspaces.length === 0">Workspace</option>
|
||||
<template x-for="workspace in $store.timeTravel.workspaces" :key="workspace.id">
|
||||
<option
|
||||
:value="workspace.id"
|
||||
:selected="workspace.id === $store.timeTravel.selectedWorkspaceId"
|
||||
x-text="$store.timeTravel.workspaceOptionLabel(workspace)"
|
||||
></option>
|
||||
</template>
|
||||
</select>
|
||||
</div>
|
||||
<div class="time-travel-workspace" :title="$store.timeTravel.workspacePath" x-text="$store.timeTravel.workspacePath || 'workspace'"></div>
|
||||
<span class="time-travel-spacer"></span>
|
||||
<button type="button" class="time-travel-icon-button" title="Snapshot" aria-label="Snapshot" @click="$store.timeTravel.manualSnapshot()" :disabled="$store.timeTravel.busy || $store.timeTravel.loading || $store.timeTravel.isLocked()">
|
||||
<span class="material-symbols-outlined">add_a_photo</span>
|
||||
</button>
|
||||
<button type="button" class="time-travel-icon-button" title="Refresh" aria-label="Refresh" @click="$store.timeTravel.refresh({ keepSelection: true })" :disabled="$store.timeTravel.loading">
|
||||
<button type="button" class="time-travel-icon-button" title="Refresh" aria-label="Refresh" @click="$store.timeTravel.refresh({ keepSelection: true, reloadWorkspaces: true })" :disabled="$store.timeTravel.loading">
|
||||
<span class="material-symbols-outlined" :class="{ spinning: $store.timeTravel.loading }">refresh</span>
|
||||
</button>
|
||||
</div>
|
||||
|
|
@ -270,7 +283,6 @@
|
|||
|
||||
.time-travel-toolbar,
|
||||
.time-travel-status,
|
||||
.time-travel-title,
|
||||
.time-travel-filter,
|
||||
.time-travel-actions,
|
||||
.time-travel-tool-button,
|
||||
|
|
@ -293,25 +305,48 @@
|
|||
background: color-mix(in srgb, var(--color-background) 91%, #000 9%);
|
||||
}
|
||||
|
||||
.time-travel-title {
|
||||
gap: 7px;
|
||||
font-weight: 750;
|
||||
font-size: 0.9rem;
|
||||
}
|
||||
|
||||
.time-travel-title .material-symbols-outlined {
|
||||
font-size: 19px;
|
||||
}
|
||||
|
||||
.time-travel-workspace {
|
||||
.time-travel-workspace-picker {
|
||||
display: flex;
|
||||
align-items: center;
|
||||
gap: 6px;
|
||||
flex: 0 1 520px;
|
||||
min-width: 0;
|
||||
max-width: 45%;
|
||||
overflow: hidden;
|
||||
max-width: 48%;
|
||||
height: 32px;
|
||||
padding: 0 8px;
|
||||
border: 1px solid color-mix(in srgb, var(--color-border) 58%, transparent);
|
||||
border-radius: 7px;
|
||||
background: color-mix(in srgb, var(--color-panel) 70%, transparent);
|
||||
}
|
||||
|
||||
.time-travel-workspace-picker .material-symbols-outlined {
|
||||
flex: 0 0 auto;
|
||||
color: var(--color-text-muted);
|
||||
font-size: 17px;
|
||||
}
|
||||
|
||||
.time-travel-workspace-picker select {
|
||||
width: 100%;
|
||||
min-width: 0;
|
||||
border: 0;
|
||||
outline: 0;
|
||||
background: transparent;
|
||||
color: var(--color-text);
|
||||
cursor: pointer;
|
||||
font-size: 0.76rem;
|
||||
overflow: hidden;
|
||||
text-overflow: ellipsis;
|
||||
white-space: nowrap;
|
||||
font-family: var(--font-family-code);
|
||||
font-size: 0.72rem;
|
||||
}
|
||||
|
||||
.time-travel-workspace-picker select:disabled {
|
||||
cursor: not-allowed;
|
||||
opacity: 0.65;
|
||||
}
|
||||
|
||||
.time-travel-workspace-picker option {
|
||||
background: var(--color-background);
|
||||
color: var(--color-text);
|
||||
}
|
||||
|
||||
.time-travel-spacer {
|
||||
|
|
@ -877,12 +912,17 @@
|
|||
justify-content: flex-start;
|
||||
}
|
||||
|
||||
.time-travel-workspace {
|
||||
display: none;
|
||||
.time-travel-workspace-picker {
|
||||
flex: 1 1 160px;
|
||||
max-width: none;
|
||||
}
|
||||
}
|
||||
|
||||
@container (max-width: 520px) {
|
||||
.time-travel-workspace-picker {
|
||||
flex-basis: 120px;
|
||||
}
|
||||
|
||||
.time-travel-tool-button span:last-child {
|
||||
display: none;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -29,10 +29,13 @@ function apiPath(name) {
|
|||
|
||||
const model = {
|
||||
loading: false,
|
||||
workspaceLoading: false,
|
||||
busy: false,
|
||||
error: "",
|
||||
payload: null,
|
||||
contextId: "",
|
||||
workspaces: [],
|
||||
selectedWorkspaceId: "",
|
||||
workspacePath: "",
|
||||
fileFilter: "",
|
||||
selectedHash: "",
|
||||
|
|
@ -63,15 +66,20 @@ const model = {
|
|||
if (this._mode !== "modal") {
|
||||
this.setupCanvasSurface(element);
|
||||
}
|
||||
this.contextId = this.resolveContextId();
|
||||
if (!this.payload && !this.loading) {
|
||||
await this.refresh({ contextId: this.contextId });
|
||||
const nextContextId = this.resolveContextId();
|
||||
const resetWorkspace = this._mode === "modal" || this.contextId !== nextContextId || !this.selectedWorkspaceId;
|
||||
this.contextId = nextContextId;
|
||||
await this.loadWorkspaces({ contextId: this.contextId, reset: resetWorkspace });
|
||||
if (this._mode === "modal" || !this.payload || resetWorkspace) {
|
||||
await this.refresh({ contextId: this.contextId, keepSelection: !resetWorkspace, skipWorkspaceLoad: true });
|
||||
}
|
||||
},
|
||||
|
||||
async onOpen(payload = {}) {
|
||||
const nextContextId = String(payload.contextId || payload.context_id || this.resolveContextId() || "");
|
||||
await this.refresh({ contextId: nextContextId });
|
||||
this.contextId = nextContextId;
|
||||
await this.loadWorkspaces({ contextId: nextContextId, reset: true });
|
||||
await this.refresh({ contextId: nextContextId, skipWorkspaceLoad: true });
|
||||
},
|
||||
|
||||
cleanup() {
|
||||
|
|
@ -106,18 +114,45 @@ const model = {
|
|||
if (this._filterTimer) clearTimeout(this._filterTimer);
|
||||
this._filterTimer = setTimeout(() => {
|
||||
this._filterTimer = null;
|
||||
this.refresh({ keepSelection: false });
|
||||
this.refresh({ keepSelection: false, skipWorkspaceLoad: true });
|
||||
}, 240);
|
||||
},
|
||||
|
||||
async loadWorkspaces(options = {}) {
|
||||
const contextId = String(options.contextId || options.context_id || this.resolveContextId() || "");
|
||||
this.workspaceLoading = true;
|
||||
try {
|
||||
const response = await callJsonApi(apiPath("history_workspaces"), {
|
||||
context_id: contextId,
|
||||
});
|
||||
if (!response?.ok) throw new Error(response?.error || "Could not load workspaces.");
|
||||
this.workspaces = Array.isArray(response.workspaces) ? response.workspaces : [];
|
||||
const defaultWorkspaceId = String(response.default_workspace_id || "");
|
||||
const hasSelected = this.workspaces.some((workspace) => workspace?.id === this.selectedWorkspaceId);
|
||||
if (options.reset || !this.selectedWorkspaceId || !hasSelected) {
|
||||
this.selectedWorkspaceId = defaultWorkspaceId || String(this.workspaces[0]?.id || "");
|
||||
}
|
||||
} catch (error) {
|
||||
this.workspaces = [];
|
||||
this.selectedWorkspaceId = "";
|
||||
this.error = error instanceof Error ? error.message : String(error);
|
||||
} finally {
|
||||
this.workspaceLoading = false;
|
||||
}
|
||||
},
|
||||
|
||||
async refresh(options = {}) {
|
||||
const contextId = String(options.contextId || options.context_id || this.resolveContextId() || "");
|
||||
if (!options.skipWorkspaceLoad && (options.reloadWorkspaces || this.workspaces.length === 0)) {
|
||||
await this.loadWorkspaces({ contextId, reset: Boolean(options.resetWorkspace) });
|
||||
}
|
||||
const seq = ++this._requestSeq;
|
||||
this.loading = true;
|
||||
this.error = "";
|
||||
try {
|
||||
const response = await callJsonApi(apiPath("history_list"), {
|
||||
context_id: contextId,
|
||||
workspace_id: this.selectedWorkspaceId,
|
||||
limit: 100,
|
||||
offset: 0,
|
||||
file_filter: this.fileFilter,
|
||||
|
|
@ -126,7 +161,17 @@ const model = {
|
|||
if (!response?.ok) throw new Error(response?.error || "Could not load history.");
|
||||
this.payload = response;
|
||||
this.contextId = String(response.context_id || contextId || "");
|
||||
this.workspacePath = String(response.workspace?.display_path || response.workspace?.path || "");
|
||||
if (response.workspace?.id && !this.selectedWorkspaceId) {
|
||||
this.selectedWorkspaceId = String(response.workspace.id);
|
||||
}
|
||||
const selectedWorkspace = this.selectedWorkspace();
|
||||
this.workspacePath = String(
|
||||
response.workspace?.display_path
|
||||
|| response.workspace?.path
|
||||
|| selectedWorkspace?.display_path
|
||||
|| selectedWorkspace?.path
|
||||
|| ""
|
||||
);
|
||||
this.reconcileSelection(Boolean(options.keepSelection));
|
||||
} catch (error) {
|
||||
if (seq !== this._requestSeq) return;
|
||||
|
|
@ -143,6 +188,7 @@ const model = {
|
|||
try {
|
||||
const response = await callJsonApi(apiPath("history_list"), {
|
||||
context_id: this.contextId,
|
||||
workspace_id: this.selectedWorkspaceId,
|
||||
limit: 100,
|
||||
offset: this.commits().length,
|
||||
file_filter: this.fileFilter,
|
||||
|
|
@ -190,6 +236,31 @@ const model = {
|
|||
return Boolean(this.payload?.workspace?.locked || this.payload?.workspace?.available === false);
|
||||
},
|
||||
|
||||
selectedWorkspace() {
|
||||
return (this.workspaces || []).find((workspace) => workspace?.id === this.selectedWorkspaceId) || null;
|
||||
},
|
||||
|
||||
workspaceOptionLabel(workspace) {
|
||||
const label = String(workspace?.label || workspace?.title || workspace?.name || "Workspace");
|
||||
const path = String(workspace?.display_path || workspace?.path || "");
|
||||
const suffix = workspace?.locked || workspace?.available === false ? " (unavailable)" : "";
|
||||
return path ? `${label} - ${path}${suffix}` : `${label}${suffix}`;
|
||||
},
|
||||
|
||||
async selectWorkspace(workspaceId) {
|
||||
const nextWorkspaceId = String(workspaceId || "");
|
||||
if (!nextWorkspaceId || nextWorkspaceId === this.selectedWorkspaceId || this.busy) return;
|
||||
this.selectedWorkspaceId = nextWorkspaceId;
|
||||
this.fileFilter = "";
|
||||
this.selectedHash = "";
|
||||
this.selectedPath = "";
|
||||
this.selectedDiff = null;
|
||||
this.diffError = "";
|
||||
this.previewOpen = false;
|
||||
this.preview = null;
|
||||
await this.refresh({ keepSelection: false, skipWorkspaceLoad: true });
|
||||
},
|
||||
|
||||
hasHistory() {
|
||||
return this.commits().length > 0;
|
||||
},
|
||||
|
|
@ -262,6 +333,7 @@ const model = {
|
|||
try {
|
||||
const response = await callJsonApi(apiPath("history_diff"), {
|
||||
context_id: this.contextId,
|
||||
workspace_id: this.selectedWorkspaceId,
|
||||
commit_hash: row.kind === "present" ? this.payload?.current_hash || "" : row.hash,
|
||||
path: file.path || file.old_path,
|
||||
mode: row.kind === "present" ? "present" : "commit",
|
||||
|
|
@ -284,6 +356,7 @@ const model = {
|
|||
try {
|
||||
const response = await callJsonApi(apiPath("history_snapshot"), {
|
||||
context_id: this.contextId,
|
||||
workspace_id: this.selectedWorkspaceId,
|
||||
trigger: "manual",
|
||||
});
|
||||
if (!response?.ok) throw new Error(response?.error || "Snapshot failed.");
|
||||
|
|
@ -309,6 +382,7 @@ const model = {
|
|||
try {
|
||||
const response = await callJsonApi(apiPath("history_preview"), {
|
||||
context_id: this.contextId,
|
||||
workspace_id: this.selectedWorkspaceId,
|
||||
operation,
|
||||
commit_hash: target.hash,
|
||||
});
|
||||
|
|
@ -340,6 +414,7 @@ const model = {
|
|||
try {
|
||||
const response = await callJsonApi(apiPath(endpoint), {
|
||||
context_id: this.contextId,
|
||||
workspace_id: this.selectedWorkspaceId,
|
||||
commit_hash: this.preview.commit_hash,
|
||||
metadata: { source: "time_travel_ui" },
|
||||
});
|
||||
|
|
|
|||
|
|
@ -48,6 +48,48 @@ def test_extract_json_root_string_returns_canonical_snapshot():
|
|||
assert extract_tools.extract_json_root_string('[{"tool_name":"response"}]') is None
|
||||
|
||||
|
||||
def test_litellm_global_kwargs_merge_defaults_and_config(monkeypatch):
|
||||
monkeypatch.setattr(
|
||||
models.settings,
|
||||
"get_settings",
|
||||
lambda: {"litellm_global_kwargs": {}},
|
||||
)
|
||||
|
||||
assert models._merge_litellm_call_kwargs({})["drop_params"] is True
|
||||
assert models._merge_litellm_call_kwargs({"temperature": 0}) == {
|
||||
"drop_params": True,
|
||||
"temperature": 0,
|
||||
}
|
||||
|
||||
monkeypatch.setattr(
|
||||
models.settings,
|
||||
"get_settings",
|
||||
lambda: {"litellm_global_kwargs": {"drop_params": "false", "timeout": "30"}},
|
||||
)
|
||||
|
||||
assert models._merge_litellm_call_kwargs({}) == {
|
||||
"drop_params": False,
|
||||
"timeout": 30,
|
||||
}
|
||||
|
||||
original_drop_params = getattr(models.litellm, "drop_params", None)
|
||||
had_timeout = hasattr(models.litellm, "timeout")
|
||||
original_timeout = getattr(models.litellm, "timeout", None)
|
||||
try:
|
||||
assert models.set_litellm_params() == {
|
||||
"drop_params": False,
|
||||
"timeout": 30,
|
||||
}
|
||||
assert models.litellm.drop_params is False
|
||||
assert models.litellm.timeout == 30
|
||||
finally:
|
||||
setattr(models.litellm, "drop_params", original_drop_params)
|
||||
if had_timeout:
|
||||
setattr(models.litellm, "timeout", original_timeout)
|
||||
elif hasattr(models.litellm, "timeout"):
|
||||
delattr(models.litellm, "timeout")
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_unified_call_stops_after_canonical_root_snapshot(monkeypatch):
|
||||
stream = _AsyncChunkStream(
|
||||
|
|
@ -61,6 +103,7 @@ async def test_unified_call_stops_after_canonical_root_snapshot(monkeypatch):
|
|||
|
||||
async def fake_acompletion(*args, **kwargs):
|
||||
assert kwargs["stream"] is True
|
||||
assert kwargs["drop_params"] is True
|
||||
return stream
|
||||
|
||||
async def fake_rate_limiter(*args, **kwargs):
|
||||
|
|
@ -68,6 +111,11 @@ async def test_unified_call_stops_after_canonical_root_snapshot(monkeypatch):
|
|||
|
||||
monkeypatch.setattr(models, "acompletion", fake_acompletion)
|
||||
monkeypatch.setattr(models, "apply_rate_limiter", fake_rate_limiter)
|
||||
monkeypatch.setattr(
|
||||
models.settings,
|
||||
"get_settings",
|
||||
lambda: {"litellm_global_kwargs": {}},
|
||||
)
|
||||
|
||||
wrapper = models.LiteLLMChatWrapper(
|
||||
model="test-model",
|
||||
|
|
|
|||
|
|
@ -373,3 +373,83 @@ def test_workspace_resolution_prefers_project_and_rejects_external_paths(monkeyp
|
|||
projects_mod.get_context_project_name = lambda _context: ""
|
||||
with pytest.raises(WorkspaceRejectedError):
|
||||
resolve_workspace("ctx", context_loader=lambda _ctxid: SimpleNamespace(id="ctx"))
|
||||
|
||||
|
||||
def test_selectable_workspaces_list_workdir_first_and_default_to_context_project(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
workspace,
|
||||
):
|
||||
root, _service = workspace
|
||||
(root / "workdir").mkdir()
|
||||
(root / "demo").mkdir()
|
||||
(root / "other").mkdir()
|
||||
|
||||
projects_mod = ModuleType("helpers.projects")
|
||||
projects_mod.get_active_projects_list = lambda: [
|
||||
{"name": "demo", "title": "Demo Project", "color": "#336699"},
|
||||
{"name": "other", "title": "Other Project", "color": ""},
|
||||
]
|
||||
projects_mod.get_context_project_name = lambda _context: "demo"
|
||||
projects_mod.get_project_folder = lambda name: str(root / name)
|
||||
settings_mod = ModuleType("helpers.settings")
|
||||
settings_mod.get_settings = lambda: {"workdir_path": str(root / "workdir")}
|
||||
|
||||
import helpers
|
||||
|
||||
monkeypatch.setitem(sys.modules, "helpers.projects", projects_mod)
|
||||
monkeypatch.setitem(sys.modules, "helpers.settings", settings_mod)
|
||||
monkeypatch.setattr(helpers, "projects", projects_mod, raising=False)
|
||||
monkeypatch.setattr(helpers, "settings", settings_mod, raising=False)
|
||||
|
||||
data = tt.list_selectable_workspaces(
|
||||
"ctx",
|
||||
context_loader=lambda _ctxid: SimpleNamespace(id="ctx"),
|
||||
)
|
||||
workspaces = data["workspaces"]
|
||||
demo_workspace = next(item for item in workspaces if item["project_name"] == "demo")
|
||||
|
||||
assert workspaces[0]["kind"] == "workdir"
|
||||
assert workspaces[0]["display_path"].endswith("/workdir")
|
||||
assert [item["project_name"] for item in workspaces[1:]] == ["demo", "other"]
|
||||
assert data["default_workspace_id"] == demo_workspace["id"]
|
||||
|
||||
resolved = resolve_workspace(
|
||||
"ctx",
|
||||
workspace_id=demo_workspace["id"],
|
||||
context_loader=lambda _ctxid: SimpleNamespace(id="ctx"),
|
||||
)
|
||||
|
||||
assert resolved.project_name == "demo"
|
||||
assert resolved.display_path == demo_workspace["display_path"]
|
||||
|
||||
with pytest.raises(WorkspaceRejectedError):
|
||||
resolve_workspace(
|
||||
"ctx",
|
||||
workspace_id="missing",
|
||||
context_loader=lambda _ctxid: SimpleNamespace(id="ctx"),
|
||||
)
|
||||
|
||||
|
||||
def test_external_workdir_workspace_option_is_locked(monkeypatch: pytest.MonkeyPatch):
|
||||
projects_mod = ModuleType("helpers.projects")
|
||||
projects_mod.get_active_projects_list = lambda: []
|
||||
projects_mod.get_context_project_name = lambda _context: ""
|
||||
settings_mod = ModuleType("helpers.settings")
|
||||
settings_mod.get_settings = lambda: {"workdir_path": "/tmp/not-a0"}
|
||||
|
||||
import helpers
|
||||
|
||||
monkeypatch.setitem(sys.modules, "helpers.projects", projects_mod)
|
||||
monkeypatch.setitem(sys.modules, "helpers.settings", settings_mod)
|
||||
monkeypatch.setattr(helpers, "projects", projects_mod, raising=False)
|
||||
monkeypatch.setattr(helpers, "settings", settings_mod, raising=False)
|
||||
|
||||
data = tt.list_selectable_workspaces("")
|
||||
workdir = data["workspaces"][0]
|
||||
|
||||
assert workdir["kind"] == "workdir"
|
||||
assert workdir["locked"] is True
|
||||
assert data["default_workspace_id"] == workdir["id"]
|
||||
|
||||
with pytest.raises(WorkspaceRejectedError):
|
||||
resolve_workspace("", workspace_id=workdir["id"])
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue