mirror of
https://github.com/HKUDS/OpenSpace.git
synced 2026-08-28 05:15:00 +00:00
54 lines
1.4 KiB
Python
54 lines
1.4 KiB
Python
"""Stable scope keys for background memory work."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
from typing import Any, Mapping
|
|
|
|
|
|
def _context_value(context: Any, key: str) -> Any:
|
|
if isinstance(context, Mapping):
|
|
return context.get(key)
|
|
return getattr(context, key, None)
|
|
|
|
|
|
def _normalize_scope_path(value: Any) -> str:
|
|
return str(Path(str(value)).expanduser().resolve())
|
|
|
|
|
|
def resolve_memory_task_scope_key(context: Any) -> str:
|
|
"""Return the stable scope key used to group memory background tasks."""
|
|
|
|
if context is None:
|
|
raise ValueError("Memory task scope requires a context.")
|
|
|
|
session_id = _context_value(context, "session_id")
|
|
if session_id:
|
|
return f"session_id:{session_id}"
|
|
|
|
session_dir = _context_value(context, "session_dir")
|
|
if session_dir:
|
|
return f"session_dir:{_normalize_scope_path(session_dir)}"
|
|
|
|
cwd = _context_value(context, "cwd")
|
|
if cwd:
|
|
return f"cwd:{_normalize_scope_path(cwd)}"
|
|
|
|
raise ValueError(
|
|
"Memory task scope requires context.session_dir, context.session_id, or context.cwd."
|
|
)
|
|
|
|
|
|
def maybe_memory_task_scope_key(context: Any | None) -> str | None:
|
|
if context is None:
|
|
return None
|
|
try:
|
|
return resolve_memory_task_scope_key(context)
|
|
except ValueError:
|
|
return None
|
|
|
|
|
|
__all__ = [
|
|
"maybe_memory_task_scope_key",
|
|
"resolve_memory_task_scope_key",
|
|
]
|