Helper to read runtime context injected via `_context` kwarg.
| 28 | |
| 29 | |
| 30 | class FileToolContext: |
| 31 | """Helper to read runtime context injected via `_context` kwarg.""" |
| 32 | |
| 33 | def __init__(self, ctx: Dict[str, Any] | None): |
| 34 | if ctx is None: |
| 35 | raise ValueError("_context is required for file tools") |
| 36 | self._ctx = ctx |
| 37 | self.attachment_store = self._require_store(ctx.get("attachment_store")) |
| 38 | self.workspace_root = self._require_workspace(ctx.get("python_workspace_root")) |
| 39 | self.session_root = self._require_session_root(ctx.get("graph_directory"), self.workspace_root) |
| 40 | |
| 41 | @staticmethod |
| 42 | def _require_store(store: Any) -> AttachmentStore: |
| 43 | if not isinstance(store, AttachmentStore): |
| 44 | raise ValueError("attachment_store missing from _context") |
| 45 | return store |
| 46 | |
| 47 | @staticmethod |
| 48 | def _require_workspace(root: Any) -> Path: |
| 49 | if root is None: |
| 50 | raise ValueError("python_workspace_root missing from _context") |
| 51 | path = Path(root).resolve() |
| 52 | path.mkdir(parents=True, exist_ok=True) |
| 53 | return path |
| 54 | |
| 55 | @staticmethod |
| 56 | def _require_session_root(root: Any, workspace_root: Path) -> Path: |
| 57 | base = root or workspace_root.parent |
| 58 | path = Path(base).resolve() |
| 59 | path.mkdir(parents=True, exist_ok=True) |
| 60 | return path |
| 61 | |
| 62 | def resolve_under_workspace(self, relative_path: str | Path) -> Path: |
| 63 | rel = Path(relative_path) |
| 64 | target = rel.resolve() if rel.is_absolute() else (self.workspace_root / rel).resolve() |
| 65 | if self.workspace_root not in target.parents and target != self.workspace_root: |
| 66 | raise ValueError("Path is outside workspace") |
| 67 | return target |
| 68 | |
| 69 | def resolve_under_session(self, relative_path: str | Path) -> Path: |
| 70 | raw = Path(relative_path) |
| 71 | candidates = [] |
| 72 | if raw.is_absolute(): |
| 73 | candidates.append(raw.resolve()) |
| 74 | else: |
| 75 | candidates.append((self.session_root / raw).resolve()) |
| 76 | candidates.append(raw.resolve()) |
| 77 | for target in candidates: |
| 78 | if self.session_root in target.parents or target == self.session_root: |
| 79 | return target |
| 80 | raise ValueError("Path is outside session directory") |
| 81 | |
| 82 | def to_session_relative(self, absolute_path: str | Path | None) -> Optional[str]: |
| 83 | if not absolute_path: |
| 84 | return None |
| 85 | target = Path(absolute_path).resolve() |
| 86 | if self.session_root in target.parents or target == self.session_root: |
| 87 | return target.relative_to(self.session_root).as_posix() |
no outgoing calls
no test coverage detected