feat(acms): add scoped backend view filtering
CI / benchmark-publish (pull_request) Has been skipped
CI / lint (pull_request) Successful in 14s
CI / build (pull_request) Successful in 15s
CI / quality (pull_request) Successful in 18s
CI / security (pull_request) Successful in 33s
CI / typecheck (pull_request) Successful in 41s
CI / unit_tests (pull_request) Successful in 2m29s
CI / integration_tests (pull_request) Successful in 3m2s
CI / docker (pull_request) Successful in 39s
CI / coverage (pull_request) Successful in 4m28s
CI / benchmark-regression (pull_request) Successful in 29m5s

Add project-resource isolation to the ACMS context tier system via
ScopedBackendView, ResourceScope, alias resolution, DAG expansion,
and enforcement hooks.

Core types (scoped_view.py): ResourceScope, ScopeViolationError,
ScopedBackendView, ScopedBackendSet, create_scoped_backend_set.

Scope resolution (scope_resolution.py): ResourceAliasResolver,
validate_project_scope, validate_resource_scope, resolve_resource_scope
with registry_lookup callback for DAG expansion.

Tier integration (scoped_tiers.py): ScopedTierMixin with get_scoped,
get_scoped_by_resource, validate_fragment_scope, store_with_scope_check,
get_scoped_metrics mixed into ContextTierService.

Backward compat: tiers.py re-exports ScopedBackendView.

Tests: 74 Behave scenarios (235 steps), 8 Robot Framework cases,
7 ASV benchmark suites. Lint and typecheck clean.

ISSUES CLOSED: #193
This commit is contained in:
2026-03-06 00:49:33 +00:00
parent b5915feb6b
commit b028c80cab
14 changed files with 3681 additions and 76 deletions
+15
View File
@@ -9,6 +9,21 @@
``plan correct``), database-backed persistence, context snapshots, and
invariant enforcement during strategize. Added acceptance criteria tags and
milestone documentation to the robot suite. (#494)
- Added scoped backend view filtering for project-resource isolation in ACMS.
`ResourceScope` holds the resolved set of resource ULIDs and project names visible
to a plan (immutable, with `include_paths`/`exclude_paths` glob filtering via
`PurePath.full_match()`). `ScopedBackendView` wraps text/vector/graph backends to
auto-inject the `scope` parameter into every query, and filters `TieredFragment`
visibility by project name and resource ID. `ScopedBackendSet` bundles scoped
views for all three backend types. `ResourceAliasResolver` translates user-facing
aliases to canonical resource ULIDs with uniqueness validation.
`resolve_resource_scope()` builds a `ResourceScope` from projects with
allowlist/denylist filtering. `validate_project_scope()` and
`validate_resource_scope()` guard against out-of-scope access. Three enforcement
hooks added to `ContextTierService`: `get_scoped_by_resource`,
`validate_fragment_scope`, `store_with_scope_check`. Includes 74 Behave BDD
scenarios (with full coverage-gap tests), 8 Robot Framework integration tests,
ASV benchmarks, and reference documentation. (#193)
- Added `builtin/plan-subplan` tool for strategy actors to emit `SUBPLAN_SPAWN` or
`SUBPLAN_PARALLEL_SPAWN` decisions. Validates payload via `SubplanPayload` (Pydantic),
applies defaults (merge strategy, max_parallel, dependencies), generates rationale text,
+270
View File
@@ -0,0 +1,270 @@
"""ASV benchmarks for ACMS Scoped Backend View filtering.
Measures the performance of:
- ResourceScope creation and containment checks
- ResourceScope path matching with include/exclude globs
- ScopedBackendView fragment visibility filtering
- ScopedBackendView backend search proxying overhead
- ScopedBackendSet creation and multi-backend search
- ResourceAliasResolver alias resolution
- resolve_resource_scope allowlist/denylist resolution
- validate_project_scope and validate_resource_scope guards
"""
from __future__ import annotations
import importlib
import sys
from pathlib import Path
# Ensure the local *source* tree is importable even when ASV has an
# older build of the package installed.
_SRC = str(Path(__file__).resolve().parents[1] / "src")
if _SRC not in sys.path:
sys.path.insert(0, _SRC)
# Force-reload so ASV picks up the source tree version.
import cleveragents # noqa: E402
importlib.reload(cleveragents)
from cleveragents.domain.models.acms.scope_resolution import ( # noqa: E402
resolve_resource_scope,
validate_project_scope,
validate_resource_scope,
)
from cleveragents.domain.models.acms.scoped_view import ( # noqa: E402
ResourceScope,
ScopedBackendView,
create_scoped_backend_set,
)
from cleveragents.domain.models.acms.stubs import ( # noqa: E402
InMemoryGraphBackend,
InMemoryTextBackend,
InMemoryVectorBackend,
)
from cleveragents.domain.models.acms.tiers import ( # noqa: E402
ContextTier,
TieredFragment,
)
from cleveragents.domain.models.core.project import ( # noqa: E402
LinkedResource,
NamespacedProject,
)
# ---------------------------------------------------------------------------
# ResourceScope benchmarks
# ---------------------------------------------------------------------------
class ResourceScopeCreationSuite:
"""Benchmark ResourceScope instantiation overhead."""
timeout = 60
def time_create_small_scope(self) -> None:
ResourceScope(
resource_ids=frozenset({"R1", "R2"}),
project_names=frozenset({"local/api"}),
)
def time_create_large_scope(self) -> None:
ResourceScope(
resource_ids=frozenset(f"R{i:04d}" for i in range(1000)),
project_names=frozenset(f"ns/proj{i}" for i in range(10)),
)
def time_contains_resource(self) -> None:
scope = ResourceScope(
resource_ids=frozenset(f"R{i:04d}" for i in range(1000)),
project_names=frozenset({"p1"}),
)
scope.contains_resource("R0500")
def time_contains_project(self) -> None:
scope = ResourceScope(
resource_ids=frozenset({"R1"}),
project_names=frozenset(f"ns/proj{i}" for i in range(100)),
)
scope.contains_project("ns/proj50")
# ---------------------------------------------------------------------------
# Path matching benchmarks
# ---------------------------------------------------------------------------
class PathMatchingSuite:
"""Benchmark ResourceScope.matches_path() with various patterns."""
timeout = 60
def setup(self) -> None:
self.scope_include = ResourceScope(
resource_ids=frozenset({"R1"}),
project_names=frozenset({"p1"}),
include_paths=("src/**/*.py", "tests/**/*.py"),
)
self.scope_exclude = ResourceScope(
resource_ids=frozenset({"R1"}),
project_names=frozenset({"p1"}),
exclude_paths=("**/node_modules/**", "**/__pycache__/**"),
)
self.scope_both = ResourceScope(
resource_ids=frozenset({"R1"}),
project_names=frozenset({"p1"}),
include_paths=("src/**/*.py",),
exclude_paths=("**/test_*",),
)
def time_match_include_hit(self) -> None:
self.scope_include.matches_path("src/models/user.py")
def time_match_include_miss(self) -> None:
self.scope_include.matches_path("docs/readme.md")
def time_match_exclude_pass(self) -> None:
self.scope_exclude.matches_path("src/main.py")
def time_match_exclude_reject(self) -> None:
self.scope_exclude.matches_path("node_modules/foo/bar.js")
def time_match_combined(self) -> None:
self.scope_both.matches_path("src/main.py")
# ---------------------------------------------------------------------------
# ScopedBackendView benchmarks
# ---------------------------------------------------------------------------
class FragmentFilteringSuite:
"""Benchmark ScopedBackendView.is_visible() fragment filtering."""
timeout = 60
def setup(self) -> None:
self.view = ScopedBackendView(
allowed_projects=frozenset({"proj-a"}),
resource_ids=frozenset({"R1", "R2"}),
)
self.frag_visible = TieredFragment(
fragment_id="f1",
content="c",
tier=ContextTier("hot"),
project_name="proj-a",
resource_id="R1",
)
self.frag_hidden = TieredFragment(
fragment_id="f2",
content="c",
tier=ContextTier("hot"),
project_name="proj-b",
resource_id="R1",
)
def time_visible_fragment(self) -> None:
self.view.is_visible(self.frag_visible)
def time_hidden_fragment(self) -> None:
self.view.is_visible(self.frag_hidden)
def time_effective_scope(self) -> None:
self.view.effective_scope()
class BackendProxyingSuite:
"""Benchmark ScopedBackendView search proxying overhead."""
timeout = 60
def setup(self) -> None:
scope = ResourceScope(
resource_ids=frozenset({"R1", "R2"}),
project_names=frozenset({"local/api"}),
)
self.view = ScopedBackendView.from_resource_scope(scope)
self.text_backend = InMemoryTextBackend()
self.vector_backend = InMemoryVectorBackend()
self.graph_backend = InMemoryGraphBackend()
self.embedding: list[float] = [0.1] * 768
def time_search_text(self) -> None:
self.view.search_text(self.text_backend, "auth flow")
def time_search_vector(self) -> None:
self.view.search_vector(self.vector_backend, self.embedding)
def time_search_graph(self) -> None:
self.view.search_graph(
self.graph_backend, "SELECT ?s WHERE { ?s a uko:Container }"
)
# ---------------------------------------------------------------------------
# ScopedBackendSet benchmarks
# ---------------------------------------------------------------------------
class BackendSetSuite:
"""Benchmark ScopedBackendSet creation and multi-backend search."""
timeout = 60
def setup(self) -> None:
self.scope = ResourceScope(
resource_ids=frozenset({"R1"}),
project_names=frozenset({"local/api"}),
)
def time_create_backend_set(self) -> None:
create_scoped_backend_set(
self.scope,
text_backend=InMemoryTextBackend(),
vector_backend=InMemoryVectorBackend(),
graph_backend=InMemoryGraphBackend(),
)
def time_backend_set_text_search(self) -> None:
bset = create_scoped_backend_set(self.scope, text_backend=InMemoryTextBackend())
bset.search_text("query")
# ---------------------------------------------------------------------------
# Scope resolution benchmarks
# ---------------------------------------------------------------------------
class ScopeResolutionSuite:
"""Benchmark resolve_resource_scope and validation guards."""
timeout = 60
def setup(self) -> None:
self.project = NamespacedProject(
name="api",
namespace="local",
linked_resources=[
LinkedResource(resource_id="01HQ8ZDRX50000000000000001", alias="repo"),
LinkedResource(resource_id="01HQ8ZDRX50000000000000002", alias="docs"),
],
)
self.scope = ResourceScope(
resource_ids=frozenset({"R1", "R2", "R3"}),
project_names=frozenset({"p1", "p2"}),
)
def time_resolve_scope_no_filter(self) -> None:
resolve_resource_scope([self.project])
def time_resolve_scope_with_include(self) -> None:
resolve_resource_scope([self.project], include_resources=("repo",))
def time_resolve_scope_with_exclude(self) -> None:
resolve_resource_scope([self.project], exclude_resources=("docs",))
def time_validate_project_scope(self) -> None:
validate_project_scope(frozenset({"p1"}), frozenset({"p1", "p2"}))
def time_validate_resource_scope(self) -> None:
validate_resource_scope(frozenset({"R1"}), self.scope)
+271
View File
@@ -0,0 +1,271 @@
# Scoped Backend View
The Scoped Backend View layer provides project-resource isolation for the
Advanced Context Management System (ACMS). It ensures that actors and
strategies never see resources outside their plan's scope, enforcing
cross-project isolation at both the backend query and fragment visibility
levels.
## Architecture
```
Plan ──► ResourceScope ──► ScopedBackendView ──► Backend Protocol ──► Physical Store
│ │
│ ├── search_text(backend, query)
│ ├── search_vector(backend, embedding)
│ ├── search_graph(backend, query)
│ └── is_visible(fragment) → bool
├── resource_ids: frozenset[str]
├── project_names: frozenset[str]
├── include_paths / exclude_paths
└── include_resources / exclude_resources
```
Each plan is assigned a `ResourceScope` at creation time. The scope is
immutable (frozen) — once created, a plan's visible resources never change.
`ScopedBackendView` wraps backend queries to automatically inject the
`scope` parameter, and `ScopedBackendSet` bundles scoped views for all
three backend types (text, vector, graph).
## ResourceScope
Resolved set of resources and projects visible to a plan. Created by
`resolve_resource_scope()` during plan creation.
Based on `docs/specification.md` lines 42510--42525.
| Field | Type | Description |
|-------|------|-------------|
| `resource_ids` | `frozenset[str]` | Resource ULIDs in scope (empty = no resources after filtering) |
| `project_names` | `frozenset[str]` | Namespaced project names in scope (min 1) |
| `temporal_scope` | `str` | One of `"current"`, `"recent"`, `"all"` (default: `"current"`) |
| `include_resources` | `tuple[str, ...]` | Resource allowlist (names/aliases). Empty = all |
| `exclude_resources` | `tuple[str, ...]` | Resource denylist. Applied after allowlist |
| `include_paths` | `tuple[str, ...]` | Path glob allowlist. Empty = all |
| `exclude_paths` | `tuple[str, ...]` | Path glob denylist. Applied after allowlist |
### Methods
| Method | Signature | Returns |
|--------|-----------|---------|
| `contains_resource` | `(resource_id: str)` | `bool` |
| `contains_project` | `(project_name: str)` | `bool` |
| `matches_path` | `(path: str)` | `bool` — applies include/exclude glob filters |
Path matching uses `pathlib.PurePath.full_match()` which supports recursive
`**` glob patterns (e.g., `src/**/*.py`).
## ScopedBackendView
Wraps a backend to restrict queries to a resource scope. Operates at two
levels: backend-level (injects `scope` into queries) and fragment-level
(filters `TieredFragment` by project name and resource ID).
Based on `docs/specification.md` lines 42548--42566.
| Field | Type | Description |
|-------|------|-------------|
| `allowed_projects` | `frozenset[str]` | Project names the actor can access (min 1) |
| `resource_ids` | `frozenset[str]` | Resource ULIDs in scope (empty = project-only filtering) |
| `denied_resource_ids` | `frozenset[str]` | Resource ULIDs explicitly denied (denylist) |
### Methods
| Method | Signature | Returns |
|--------|-----------|---------|
| `is_resource_in_scope` | `(resource_id: str)` | `bool` |
| `effective_scope` | `()` | `frozenset[str]``resource_ids` minus `denied_resource_ids` |
| `is_visible` | `(fragment: object)` | `bool` — fragment-level project+resource check |
| `search_text` | `(backend: TextBackend, query: str, *, max_results: int = 20)` | `list[TextResult]` |
| `search_vector` | `(backend: VectorBackend, embedding: list[float], *, top_k: int = 20)` | `list[VectorResult]` |
| `search_graph` | `(backend: GraphBackend, query: str)` | `GraphResult` |
| `from_resource_scope` | `(scope: ResourceScope)` (classmethod) | `ScopedBackendView` |
**Preconditions:**
- `allowed_projects` must be non-empty.
- When `resource_ids` is populated, search methods raise `ScopeViolationError` if
`effective_scope()` is empty (all resources denied).
**Postconditions:**
- Search results are restricted to in-scope resources via the `scope` parameter.
- Fragments with empty `project_name` are always excluded.
## ScopedBackendSet
Bundle of scoped backend views for text, vector, and graph backends.
Created by `create_scoped_backend_set()` during initial context assembly.
| Field | Type | Description |
|-------|------|-------------|
| `view` | `ScopedBackendView` | The scoped view enforcing isolation |
| `text_backend` | `TextBackend \| None` | Text search backend (None if unavailable) |
| `vector_backend` | `VectorBackend \| None` | Vector search backend (None if unavailable) |
| `graph_backend` | `GraphBackend \| None` | Graph query backend (None if unavailable) |
### Methods
| Method | Signature | Returns |
|--------|-----------|---------|
| `search_text` | `(query: str, *, max_results: int = 20)` | `list[TextResult]` — empty if no backend |
| `search_vector` | `(embedding: list[float], *, top_k: int = 20)` | `list[VectorResult]` — empty if no backend |
| `search_graph` | `(query: str)` | `GraphResult` — empty if no backend |
## ScopeViolationError
Exception raised when a context request references resources outside scope.
| Attribute | Type | Description |
|-----------|------|-------------|
| `resource_ids` | `tuple[str, ...]` | Out-of-scope resource identifiers |
| `scope_project_names` | `tuple[str, ...]` | Project names that define the scope |
## ResourceAliasResolver
Stateless helper that resolves resource aliases to canonical resource ULIDs.
Used during scope resolution to translate user-facing resource names.
| Method | Signature | Returns |
|--------|-----------|---------|
| `resolve` | `(project: NamespacedProject, alias_or_id: str)` | `str \| None` |
| `resolve_many` | `(project: NamespacedProject, names: tuple[str, ...])` | `frozenset[str]` |
| `validate_uniqueness` | `(project: NamespacedProject)` | `list[str]` — error messages for duplicates |
## Factory and Validation Functions
### `create_scoped_backend_set()`
```python
def create_scoped_backend_set(
scope: ResourceScope,
*,
text_backend: TextBackend | None = None,
vector_backend: VectorBackend | None = None,
graph_backend: GraphBackend | None = None,
) -> ScopedBackendSet: ...
```
### `resolve_resource_scope()`
```python
def resolve_resource_scope(
projects: list[NamespacedProject],
*,
include_resources: tuple[str, ...] = (),
exclude_resources: tuple[str, ...] = (),
include_paths: tuple[str, ...] = (),
exclude_paths: tuple[str, ...] = (),
temporal_scope: str = "current",
registry_lookup: Callable[[str], frozenset[str]] | None = None,
) -> ResourceScope: ...
```
Collects all resource IDs from linked resources across projects, then applies
allowlist/denylist filtering using alias resolution. When `registry_lookup` is
provided, each resource ID is expanded to include its DAG descendants (per
spec §42517--42521).
### `validate_project_scope()`
```python
def validate_project_scope(
requested_projects: frozenset[str],
available_projects: frozenset[str],
) -> None: ... # raises ScopeViolationError
```
### `validate_resource_scope()`
```python
def validate_resource_scope(
requested_resource_ids: frozenset[str],
scope: ResourceScope,
) -> None: ... # raises ScopeViolationError
```
## Usage Examples
### Basic Scope Resolution
```python
from cleveragents.domain.models.acms.scope_resolution import resolve_resource_scope
# Resolve scope from a project — includes all linked resources
scope = resolve_resource_scope([project])
# Allowlist: only include the "repo" resource (by alias)
scope = resolve_resource_scope([project], include_resources=("repo",))
# Denylist: exclude "docs" resource, keep everything else
scope = resolve_resource_scope([project], exclude_resources=("docs",))
# Combined: include only Python files, exclude test files
scope = resolve_resource_scope(
[project],
include_paths=("src/**/*.py",),
exclude_paths=("**/test_*", "**/*_test.py"),
)
```
### Scoped Backend Queries
```python
from cleveragents.domain.models.acms.scoped_view import (
ScopedBackendView,
create_scoped_backend_set,
)
# Create a scoped view from a resolved scope
view = ScopedBackendView.from_resource_scope(scope)
# Search with automatic scope injection
results = view.search_text(text_backend, "authentication flow")
vectors = view.search_vector(vector_backend, embedding)
graph = view.search_graph(graph_backend, "SELECT ?s WHERE { ?s a uko:Class }")
# Or use ScopedBackendSet for convenience
backend_set = create_scoped_backend_set(
scope,
text_backend=text_backend,
vector_backend=vector_backend,
graph_backend=graph_backend,
)
results = backend_set.search_text("authentication flow")
```
### Scope Validation Guards
```python
from cleveragents.domain.models.acms.scope_resolution import (
validate_project_scope,
validate_resource_scope,
)
from cleveragents.domain.models.acms.scoped_view import ScopeViolationError
try:
validate_project_scope(
requested_projects=frozenset({"local/api", "local/frontend"}),
available_projects=frozenset({"local/api"}),
)
except ScopeViolationError as e:
# "Projects not in scope: local/frontend. Available: local/api"
print(e)
```
## Module Locations
| Module | Purpose |
|--------|---------|
| `cleveragents.domain.models.acms.scoped_view` | Core types: `ResourceScope`, `ScopedBackendView`, `ScopedBackendSet` |
| `cleveragents.domain.models.acms.scope_resolution` | Resolution: `ResourceAliasResolver`, `resolve_resource_scope`, validation guards |
| `cleveragents.domain.models.acms.tiers` | Re-exports `ScopedBackendView` for backward compat |
| `cleveragents.domain.models.acms` | Package-level re-exports of all public symbols |
| `cleveragents.application.services.context_tiers` | Enforcement hooks (`get_scoped`, `get_scoped_by_resource`, etc.) |
## Related
- [ACMS Backends Reference](acms_backends.md) — Backend protocol definitions
- [Specification: ACMS > Scoped Views](../specification.md) — Authoritative design (lines 42501--42566)
- [Forgejo Issue #193](https://git.cleverthis.com/cleveragents/cleveragents-core/issues/193) — Feature request
+465
View File
@@ -0,0 +1,465 @@
Feature: Scoped Backend View Filtering
As a developer
I want scoped backend views that filter text/vector/graph queries to project resources
So that plans never see resources outside their scope (cross-project isolation)
# ---- ResourceScope ----
Scenario: Create a valid ResourceScope
Given sv a ResourceScope with resources "RES01", "RES02" and projects "local/api"
Then sv the scope should contain resource "RES01"
And sv the scope should contain resource "RES02"
And sv the scope should not contain resource "RES99"
And sv the scope should contain project "local/api"
And sv the scope should not contain project "local/frontend"
Scenario: ResourceScope is frozen
Given sv a ResourceScope with resources "RES01" and projects "local/api"
Then sv modifying the scope should raise an error
Scenario: ResourceScope temporal_scope defaults to current
Given sv a ResourceScope with resources "RES01" and projects "local/api"
Then sv temporal_scope should be "current"
Scenario: ResourceScope rejects invalid temporal_scope
Then sv creating a scope with temporal_scope "invalid" should raise ValidationError
Scenario: ResourceScope allows empty resource_ids after filtering
Then sv creating a scope with no resources should succeed
Scenario: ResourceScope rejects empty project_names
Then sv creating a scope with no projects should raise ValidationError
# ---- ResourceScope path filtering ----
Scenario: ResourceScope matches_path with include patterns
Given sv a ResourceScope with include_paths "src/**/*.py", "tests/**/*.py"
Then sv path "src/main.py" should match
And sv path "tests/test_foo.py" should match
And sv path "docs/readme.md" should not match
Scenario: ResourceScope matches_path with exclude patterns
Given sv a ResourceScope with exclude_paths "**/node_modules/**", "**/__pycache__/**"
Then sv path "src/main.py" should match
And sv path "node_modules/foo/bar.js" should not match
And sv path "src/__pycache__/foo.pyc" should not match
Scenario: ResourceScope matches_path with both include and exclude
Given sv a ResourceScope with include_paths "src/**/*.py" and exclude_paths "**/test_*"
Then sv path "src/main.py" should match
And sv path "src/test_main.py" should not match
And sv path "docs/readme.md" should not match
# ---- ScopedBackendView: fragment-level filtering ----
Scenario: ScopedBackendView filters fragments by project name
Given sv a project-only ScopedBackendView for "proj-a"
And sv a fragment "frag-a" in project "proj-a" with resource "RES01"
And sv a fragment "frag-b" in project "proj-b" with resource "RES02"
Then sv fragment "frag-a" should be visible
And sv fragment "frag-b" should not be visible
Scenario: ScopedBackendView excludes fragments with empty project_name
Given sv a project-only ScopedBackendView for "proj-a"
And sv a fragment "frag-x" with no project
Then sv fragment "frag-x" should not be visible
Scenario: ScopedBackendView filters by resource_ids when set
Given sv a project-and-resource ScopedBackendView for "proj-a" resources "RES01"
And sv a fragment "frag-a" in project "proj-a" with resource "RES01"
And sv a fragment "frag-c" in project "proj-a" with resource "RES99"
Then sv fragment "frag-a" should be visible
And sv fragment "frag-c" should not be visible
Scenario: ScopedBackendView denied_resource_ids blocks specific resources
Given sv a deny-resource ScopedBackendView for "proj-a" allow "RES01", "RES02" deny "RES02"
And sv a fragment "frag-a" in project "proj-a" with resource "RES01"
And sv a fragment "frag-d" in project "proj-a" with resource "RES02"
Then sv fragment "frag-a" should be visible
And sv fragment "frag-d" should not be visible
# ---- ScopedBackendView: backend-level search proxying ----
Scenario: ScopedBackendView proxies text search with scope injected
Given sv a ResourceScope with resources "RES01", "RES02" and projects "local/api"
And sv a ScopedBackendView from that scope
And sv an InMemoryTextBackend
When sv I search text for "auth flow"
Then sv the text search should return an empty list
And sv the text search should not raise an error
Scenario: ScopedBackendView proxies vector search with scope injected
Given sv a ResourceScope with resources "RES01" and projects "local/api"
And sv a ScopedBackendView from that scope
And sv an InMemoryVectorBackend
When sv I search vector with embedding
Then sv the vector search should return an empty list
Scenario: ScopedBackendView proxies graph search with scope injected
Given sv a ResourceScope with resources "RES01" and projects "local/api"
And sv a ScopedBackendView from that scope
And sv an InMemoryGraphBackend
When sv I search graph for "SELECT ?s WHERE { ?s a uko:Container }"
Then sv the graph search should return an empty GraphResult
# ---- ScopedBackendView: effective_scope ----
Scenario: effective_scope returns resource_ids minus denied
Given sv a triple-resource ScopedBackendView for "p" with "R1", "R2", "R3" denying "R2"
Then sv effective_scope should contain "R1" and "R3" but not "R2"
Scenario: effective_scope is empty when all resources denied
Given sv a single-deny ScopedBackendView for "p" allow "R1" deny "R1"
Then sv effective_scope should be empty
# ---- ScopedBackendView: from_resource_scope ----
Scenario: from_resource_scope creates a view with scope's IDs and projects
Given sv a ResourceScope with two resources "RES01", "RES02" and two projects "local/api", "local/web"
When sv I create a ScopedBackendView from that ResourceScope
Then sv the view should allow project "local/api"
And sv the view should allow project "local/web"
And sv the view resource_ids should contain "RES01" and "RES02"
# ---- ScopedBackendSet ----
Scenario: ScopedBackendSet wraps multiple backends
Given sv a ResourceScope with resources "RES01" and projects "local/api"
And sv a ScopedBackendSet with text, vector, and graph backends
When sv I search text via backend set for "test query"
Then sv the backend set text search should succeed
Scenario: ScopedBackendSet with no text backend returns empty list
Given sv a ResourceScope with resources "RES01" and projects "local/api"
And sv a ScopedBackendSet with no text backend
When sv I search text via backend set for "test query"
Then sv the backend set text search should return empty list
# ---- ScopeViolationError ----
Scenario: ScopeViolationError carries resource and project info
When sv I raise a ScopeViolationError with resource "RES01" and projects "proj-a"
Then sv the error resource_ids should contain "RES01"
And sv the error scope_project_names should contain "proj-a"
# ---- validate_project_scope ----
Scenario: validate_project_scope passes for valid projects
Then sv validating projects "proj-a" against available "proj-a", "proj-b" should pass
Scenario: validate_project_scope rejects out-of-scope projects
Then sv validating projects "proj-c" against available "proj-a", "proj-b" should raise ScopeViolationError
Scenario: validate_project_scope rejects empty requested projects
Then sv validating empty requested projects should raise ValueError
# ---- validate_resource_scope ----
Scenario: validate_resource_scope passes for in-scope resources
Given sv a ResourceScope with resources "RES01", "RES02" and projects "local/api"
Then sv validating resources "RES01" against that scope should pass
Scenario: validate_resource_scope rejects out-of-scope resources
Given sv a ResourceScope with resources "RES01" and projects "local/api"
Then sv validating resources "RES99" against that scope should raise ScopeViolationError
# ---- ResourceAliasResolver ----
Scenario: Resolve alias to resource_id
Given sv a project "local/myproj" with resource "01HQ8ZDRX50000000000000001" aliased "repo"
And sv a ResourceAliasResolver
When sv I resolve alias "repo"
Then sv the resolved resource_id should be "01HQ8ZDRX50000000000000001"
Scenario: Resolve by resource_id directly
Given sv a project "local/myproj" with resource "01HQ8ZDRX50000000000000001" aliased "repo"
And sv a ResourceAliasResolver
When sv I resolve alias "01HQ8ZDRX50000000000000001"
Then sv the resolved resource_id should be "01HQ8ZDRX50000000000000001"
Scenario: Resolve unknown alias returns None
Given sv a project "local/myproj" with resource "01HQ8ZDRX50000000000000001" aliased "repo"
And sv a ResourceAliasResolver
When sv I resolve alias "nonexistent"
Then sv the resolved resource_id should be None
Scenario: Resolve rejects empty input
Given sv a project "local/myproj" with resource "01HQ8ZDRX50000000000000001" aliased "repo"
And sv a ResourceAliasResolver
Then sv resolving empty alias should raise ValueError
Scenario: resolve_many resolves multiple aliases
Given sv a project "local/myproj" with two resources "01HQ8ZDRX50000000000000001" aliased "repo" and "01HQ8ZDRX50000000000000002" aliased "docs"
And sv a ResourceAliasResolver
When sv I resolve many aliases "repo", "docs"
Then sv the resolved set should contain "01HQ8ZDRX50000000000000001" and "01HQ8ZDRX50000000000000002"
Scenario: validate_uniqueness detects duplicate aliases
Given sv a project "local/myproj" with duplicate alias "repo" on "01HQ8ZDRX50000000000000001" and "01HQ8ZDRX50000000000000002"
And sv a ResourceAliasResolver
When sv I validate alias uniqueness
Then sv there should be 1 uniqueness error mentioning "repo"
Scenario: validate_uniqueness passes for unique aliases
Given sv a project "local/myproj" with two resources "01HQ8ZDRX50000000000000001" aliased "repo" and "01HQ8ZDRX50000000000000002" aliased "docs"
And sv a ResourceAliasResolver
When sv I validate alias uniqueness
Then sv there should be 0 uniqueness errors
# ---- resolve_resource_scope ----
Scenario: resolve_resource_scope collects all resources from projects
Given sv a project "local/api" with linked resources "01HQ8ZDRX50000000000000001", "01HQ8ZDRX50000000000000002"
When sv I resolve the resource scope
Then sv the scope should have 2 resource_ids
And sv the scope should contain resource "01HQ8ZDRX50000000000000001"
And sv the scope should contain resource "01HQ8ZDRX50000000000000002"
Scenario: resolve_resource_scope applies include allowlist
Given sv a project "local/api" with two resources "01HQ8ZDRX50000000000000001" aliased "repo" and "01HQ8ZDRX50000000000000002" aliased "docs"
When sv I resolve the resource scope with include "repo"
Then sv the scope should have 1 resource_ids
And sv the scope should contain resource "01HQ8ZDRX50000000000000001"
Scenario: resolve_resource_scope applies exclude denylist
Given sv a project "local/api" with two resources "01HQ8ZDRX50000000000000001" aliased "repo" and "01HQ8ZDRX50000000000000002" aliased "docs"
When sv I resolve the resource scope with exclude "docs"
Then sv the scope should have 1 resource_ids
And sv the scope should contain resource "01HQ8ZDRX50000000000000001"
Scenario: resolve_resource_scope rejects empty projects
Then sv resolving scope with empty projects should raise ValueError
# ---- ContextTierService: scope enforcement ----
Scenario: get_scoped_by_resource filters by ResourceScope
Given sv a ContextTierService
And sv a ResourceScope with resources "RES01" and projects "proj-a"
When sv I store fragment "f1" project "proj-a" resource "RES01" tier "hot"
And sv I store fragment "f2" project "proj-b" resource "RES02" tier "hot"
Then sv get_scoped_by_resource should return 1 fragment
Scenario: store_with_scope_check rejects out-of-scope resource
Given sv a ContextTierService
And sv a ResourceScope with resources "RES01" and projects "proj-a"
Then sv storing fragment with resource "RES99" should raise ScopeViolationError
Scenario: store_with_scope_check rejects out-of-scope project
Given sv a ContextTierService
And sv a ResourceScope with resources "RES01" and projects "proj-a"
Then sv storing fragment with project "evil-proj" and resource "RES01" should raise ScopeViolationError
Scenario: validate_fragment_scope passes for in-scope fragment
Given sv a ContextTierService
And sv a ResourceScope with resources "RES01" and projects "proj-a"
When sv I store fragment "f3" project "proj-a" resource "RES01" tier "hot"
Then sv validating fragment "f3" against the scope should pass
Scenario: validate_fragment_scope rejects out-of-scope fragment
Given sv a ContextTierService
And sv a ResourceScope with resources "RES01" and projects "proj-a"
When sv I store fragment "f4" project "proj-b" resource "RES02" tier "hot"
Then sv validating fragment "f4" against the scope should raise ScopeViolationError
# ---- ScopedBackendSet: vector and graph ----
Scenario: ScopedBackendSet proxies vector search
Given sv a ResourceScope with resources "RES01" and projects "local/api"
And sv a ScopedBackendSet with text, vector, and graph backends
When sv I search vector via backend set
Then sv the backend set vector search should return empty list
Scenario: ScopedBackendSet proxies graph search
Given sv a ResourceScope with resources "RES01" and projects "local/api"
And sv a ScopedBackendSet with text, vector, and graph backends
When sv I search graph via backend set for "SELECT ?s WHERE { ?s a uko:X }"
Then sv the backend set graph search should return empty GraphResult
# ---- Mixed project scope guard ----
Scenario: Mixed project scope guard catches unlinked projects
Then sv validating mixed projects "proj-a", "proj-c" against available "proj-a", "proj-b" should raise ScopeViolationError
# ---- Coverage: field validator coercion (scoped_view L226, L237) ----
Scenario: ScopedBackendView coerces list resource_ids to frozenset
Then sv constructing view with list resource_ids should succeed
Scenario: ScopedBackendView coerces list denied_resource_ids to frozenset
Then sv constructing view with list denied_resource_ids should succeed
# ---- Coverage: effective_scope project-only (scoped_view L273) ----
Scenario: effective_scope returns empty for project-only view
Given sv a project-only ScopedBackendView for "proj-a"
Then sv effective_scope should be empty
# ---- Coverage: _require_effective_scope raises (scoped_view L326-330) ----
Scenario: search_text raises when all resources are denied
Given sv a single-deny ScopedBackendView for "p" allow "R1" deny "R1"
And sv an InMemoryTextBackend
Then sv searching text via denied view should raise ScopeViolationError
# ---- Coverage: ScopedBackendSet None backends (scoped_view L489, L505) ----
Scenario: ScopedBackendSet search_vector with no vector backend
Given sv a ResourceScope with resources "RES01" and projects "local/api"
And sv a ScopedBackendSet with no vector backend
Then sv searching vector via no-vector set should return empty list
Scenario: ScopedBackendSet search_graph with no graph backend
Given sv a ResourceScope with resources "RES01" and projects "local/api"
And sv a ScopedBackendSet with no graph backend
Then sv searching graph via no-graph set should return empty GraphResult
# ---- Coverage: resolve_many unresolvable alias (scoped_view L617) ----
Scenario: resolve_many skips unresolvable aliases
Given sv a project "local/myproj" with resource "01HQ8ZDRX50000000000000001" aliased "repo"
And sv a ResourceAliasResolver
When sv I resolve many aliases "repo", "nonexistent"
Then sv the resolved set size should be 1
And sv the resolved set should contain "01HQ8ZDRX50000000000000001"
# ---- Coverage: validate_uniqueness no-alias branch (scoped_view L638) ----
Scenario: validate_uniqueness ignores resources without aliases
Given sv a project with aliased and unaliased resources
And sv a ResourceAliasResolver
When sv I validate alias uniqueness
Then sv there should be 0 uniqueness errors
# ---- Coverage: validate_project_scope empty available (scoped_view L676) ----
Scenario: validate_project_scope rejects empty available projects
Then sv validating projects against empty available should raise ValueError
# ---- Coverage: validate_resource_scope empty request (scoped_view L710) ----
Scenario: validate_resource_scope passes for empty requested resources
Given sv a ResourceScope with resources "RES01" and projects "local/api"
Then sv validating empty resources against scope should pass
# ---- Coverage: resolve_resource_scope empty after filter (scoped_view L790) ----
Scenario: resolve_resource_scope produces empty after include filtering
Given sv a project "local/api" with two resources "01HQ8ZDRX50000000000000001" aliased "repo" and "01HQ8ZDRX50000000000000002" aliased "docs"
When sv I resolve the resource scope with include "nonexistent"
Then sv the scope should have 0 resource_ids
# ---- Coverage: context_tiers store empty id (L97) ----
Scenario: store rejects empty fragment_id
Given sv a ContextTierService
Then sv storing fragment with empty id should raise ValueError
# ---- Coverage: context_tiers get_scoped_view empty (L191) ----
Scenario: get_scoped_view with empty project_names returns empty
Given sv a ContextTierService
Then sv get_scoped_view with empty projects should return empty list
# ---- Coverage: context_tiers evict_lru empty store (L269) ----
Scenario: evict_lru on empty tier returns empty list
Given sv a ContextTierService
Then sv evict_lru on empty cold tier should return empty list
# ---- Coverage: context_tiers get_scoped_metrics (L311-316) ----
Scenario: get_scoped_metrics counts fragments by project
Given sv a ContextTierService
When sv I store fragment "m1" project "proj-a" resource "RES01" tier "hot"
And sv I store fragment "m2" project "proj-b" resource "RES02" tier "warm"
Then sv get_scoped_metrics for "proj-a" should show 1 hot and 0 warm
# ---- Coverage: context_tiers get_scoped_by_resource 0 rejected (L355) ----
Scenario: get_scoped_by_resource returns all when none rejected
Given sv a ContextTierService
And sv a ResourceScope with resources "RES01" and projects "proj-a"
When sv I store fragment "g1" project "proj-a" resource "RES01" tier "hot"
Then sv get_scoped_by_resource should return 1 fragment
# ---- Coverage: context_tiers validate_fragment_scope not found (L382) ----
Scenario: validate_fragment_scope rejects non-existent fragment
Given sv a ContextTierService
And sv a ResourceScope with resources "RES01" and projects "proj-a"
Then sv validating non-existent fragment should raise ValueError
# ---- Coverage: context_tiers store_with_scope_check happy path (L434) ----
Scenario: store_with_scope_check stores in-scope fragment successfully
Given sv a ContextTierService
And sv a ResourceScope with resources "RES01" and projects "proj-a"
Then sv store_with_scope_check should accept in-scope fragment
# ---- Coverage: context_tiers budget property (L443) ----
Scenario: tier budget property returns TierBudget
Given sv a ContextTierService
Then sv tier service budget should be a TierBudget
# ---- Coverage: context_tiers get_for_actor no project filter (L161) ----
Scenario: get_for_actor without project filter returns fragments
Given sv a ContextTierService
When sv I store fragment "a1" project "proj-a" resource "RES01" tier "hot"
Then sv get_for_actor without projects should return a list
# ---- Coverage: context_tiers _summarize_for_cold false branch (L481) ----
Scenario: demote warm fragment with short content to cold
Given sv a ContextTierService
When sv I store a short warm fragment "sf1"
Then sv demoting "sf1" to cold should keep content intact
# ---- P0-S1: is_visible denies empty resource_id when resource_ids set ----
Scenario: is_visible denies fragment with empty resource_id when resource filtering active
Given sv a ScopedBackendView for projects "proj-a" with resource_ids "RES01"
When sv I check visibility of a fragment with project "proj-a" and empty resource_id
Then sv the fragment should NOT be visible
Scenario: is_visible allows fragment with empty resource_id when no resource filtering
Given sv a ScopedBackendView for project "proj-a"
When sv I check visibility of a fragment with project "proj-a" and empty resource_id
Then sv the fragment should be visible
# ---- P0-S2: get_scoped enforces scope ----
Scenario: get_scoped returns fragment when in scope
Given sv a ContextTierService
And sv a ResourceScope with resources "RES01" and projects "proj-a"
When sv I store fragment "f1" project "proj-a" resource "RES01" tier "hot"
Then sv get_scoped for "f1" should return the fragment
Scenario: get_scoped returns None for out-of-scope fragment
Given sv a ContextTierService
And sv a ResourceScope with resources "RES01" and projects "proj-a"
When sv I store fragment "f2" project "proj-b" resource "RES02" tier "hot"
Then sv get_scoped for "f2" should return None
# ---- P1-SPEC1: DAG expansion via registry_lookup ----
Scenario: resolve_resource_scope expands child resources via registry_lookup
Given sv a project "local/api" with two resources "01HQ8ZDRX50000000000000001" aliased "repo" and "01HQ8ZDRX50000000000000002" aliased "docs"
When sv I resolve scope with registry_lookup expanding "01HQ8ZDRX50000000000000001" to "01HQ8ZDRX50000000000000099"
Then sv the scope should have 3 resource_ids
And sv the scope should contain resource "01HQ8ZDRX50000000000000099"
# ---- P2-B2: tuple coercion in field validators ----
Scenario: ScopedBackendView accepts tuple for allowed_projects
When sv I create a ScopedBackendView with tuple allowed_projects
Then sv the view should have 1 allowed project
# ---- P0-S2: get_scoped with empty resource_id denied ----
Scenario: get_scoped denies fragment with empty resource_id when scope has resources
Given sv a ContextTierService
And sv a ResourceScope with resources "RES01" and projects "proj-a"
When sv I store fragment "f3" with project "proj-a" and empty resource_id in tier "hot"
Then sv get_scoped for "f3" should return None
+920
View File
@@ -0,0 +1,920 @@
"""Step definitions for scoped backend view filtering.
All steps prefixed with ``sv`` to avoid AmbiguousStep collisions with
existing step files in the project.
"""
from __future__ import annotations
from typing import Any
from behave import given, then, when # type: ignore[import-untyped]
from cleveragents.domain.models.acms.scope_resolution import (
ResourceAliasResolver,
resolve_resource_scope,
validate_project_scope,
validate_resource_scope,
)
from cleveragents.domain.models.acms.scoped_view import (
ResourceScope,
ScopedBackendView,
ScopeViolationError,
create_scoped_backend_set,
)
from cleveragents.domain.models.acms.stubs import (
InMemoryGraphBackend,
InMemoryTextBackend,
InMemoryVectorBackend,
)
from cleveragents.domain.models.acms.tiers import (
ContextTier,
TieredFragment,
)
__all__: list[str] = []
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _make_fragment(
fid: str,
project: str = "",
resource: str = "",
tier: str = "hot",
) -> TieredFragment:
"""Create a TieredFragment for testing."""
return TieredFragment(
fragment_id=fid,
content=f"content-{fid}",
tier=ContextTier(tier),
project_name=project,
resource_id=resource,
)
def _make_project(
name: str,
namespace: str = "local",
links: list[tuple[str, str | None]] | None = None,
) -> Any:
"""Create a NamespacedProject with linked resources."""
from cleveragents.domain.models.core.project import (
LinkedResource,
NamespacedProject,
)
linked: list[LinkedResource] = []
if links:
for resource_id, alias in links:
linked.append(LinkedResource(resource_id=resource_id, alias=alias))
return NamespacedProject(
name=name,
namespace=namespace,
linked_resources=linked,
)
# ---------------------------------------------------------------------------
# ResourceScope
# ---------------------------------------------------------------------------
@given('sv a ResourceScope with resources "{res1}", "{res2}" and projects "{proj}"')
def step_sv_scope_two_res_one_proj(
context: Any, res1: str, res2: str, proj: str
) -> None:
context.sv_scope = ResourceScope(
resource_ids=frozenset({res1, res2}),
project_names=frozenset({proj}),
)
@given('sv a ResourceScope with resources "{res1}" and projects "{proj}"')
def step_sv_scope_one_res_one_proj(context: Any, res1: str, proj: str) -> None:
context.sv_scope = ResourceScope(
resource_ids=frozenset({res1}),
project_names=frozenset({proj}),
)
@then('sv the scope should contain resource "{res}"')
def step_sv_scope_contains_res(context: Any, res: str) -> None:
assert context.sv_scope.contains_resource(res), (
f"Expected scope to contain resource '{res}'"
)
@then('sv the scope should not contain resource "{res}"')
def step_sv_scope_not_contains_res(context: Any, res: str) -> None:
assert not context.sv_scope.contains_resource(res), (
f"Expected scope NOT to contain resource '{res}'"
)
@then('sv the scope should contain project "{proj}"')
def step_sv_scope_contains_proj(context: Any, proj: str) -> None:
assert context.sv_scope.contains_project(proj)
@then('sv the scope should not contain project "{proj}"')
def step_sv_scope_not_contains_proj(context: Any, proj: str) -> None:
assert not context.sv_scope.contains_project(proj)
@then("sv modifying the scope should raise an error")
def step_sv_scope_frozen(context: Any) -> None:
from pydantic import ValidationError
try:
context.sv_scope.resource_ids = frozenset({"NEW"})
raise AssertionError("Expected frozen model error")
except (ValidationError, AttributeError, TypeError):
pass
@then('sv temporal_scope should be "{expected}"')
def step_sv_temporal_scope(context: Any, expected: str) -> None:
assert context.sv_scope.temporal_scope == expected
@then('sv creating a scope with temporal_scope "{ts}" should raise ValidationError')
def step_sv_scope_bad_temporal(context: Any, ts: str) -> None:
from pydantic import ValidationError
try:
ResourceScope(
resource_ids=frozenset({"R1"}),
project_names=frozenset({"p1"}),
temporal_scope=ts,
)
raise AssertionError("Expected ValidationError")
except ValidationError:
pass
@then("sv creating a scope with no resources should succeed")
def step_sv_scope_empty_resources(context: Any) -> None:
scope = ResourceScope(
resource_ids=frozenset(),
project_names=frozenset({"p1"}),
)
assert len(scope.resource_ids) == 0
assert scope.contains_project("p1")
@then("sv creating a scope with no projects should raise ValidationError")
def step_sv_scope_no_projects(context: Any) -> None:
from pydantic import ValidationError
try:
ResourceScope(
resource_ids=frozenset({"R1"}),
project_names=frozenset(),
)
raise AssertionError("Expected ValidationError")
except ValidationError:
pass
# ---------------------------------------------------------------------------
# ResourceScope path filtering
# ---------------------------------------------------------------------------
@given('sv a ResourceScope with include_paths "{p1}", "{p2}"')
def step_sv_scope_include_paths(context: Any, p1: str, p2: str) -> None:
context.sv_scope = ResourceScope(
resource_ids=frozenset({"R1"}),
project_names=frozenset({"p1"}),
include_paths=(p1, p2),
)
@given('sv a ResourceScope with exclude_paths "{p1}", "{p2}"')
def step_sv_scope_exclude_paths(context: Any, p1: str, p2: str) -> None:
context.sv_scope = ResourceScope(
resource_ids=frozenset({"R1"}),
project_names=frozenset({"p1"}),
exclude_paths=(p1, p2),
)
@given('sv a ResourceScope with include_paths "{inc}" and exclude_paths "{exc}"')
def step_sv_scope_inc_exc_paths(context: Any, inc: str, exc: str) -> None:
context.sv_scope = ResourceScope(
resource_ids=frozenset({"R1"}),
project_names=frozenset({"p1"}),
include_paths=(inc,),
exclude_paths=(exc,),
)
@then('sv path "{path}" should match')
def step_sv_path_matches(context: Any, path: str) -> None:
assert context.sv_scope.matches_path(path), (
f"Expected path '{path}' to match scope filters"
)
@then('sv path "{path}" should not match')
def step_sv_path_not_matches(context: Any, path: str) -> None:
assert not context.sv_scope.matches_path(path), (
f"Expected path '{path}' NOT to match scope filters"
)
# ---------------------------------------------------------------------------
# ScopedBackendView: fragment-level filtering
# ---------------------------------------------------------------------------
@given('sv a project-only ScopedBackendView for "{proj}"')
def step_sv_view_one_proj(context: Any, proj: str) -> None:
context.sv_view = ScopedBackendView(
allowed_projects=frozenset({proj}),
)
context.sv_fragments = {} # dict[str, TieredFragment]
@given('sv a project-and-resource ScopedBackendView for "{proj}" resources "{res}"')
def step_sv_view_proj_res(context: Any, proj: str, res: str) -> None:
context.sv_view = ScopedBackendView(
allowed_projects=frozenset({proj}),
resource_ids=frozenset({res}),
)
context.sv_fragments = {}
@given(
'sv a deny-resource ScopedBackendView for "{proj}" allow "{r1}", "{r2}" deny "{d1}"'
)
def step_sv_view_proj_res_denied(
context: Any, proj: str, r1: str, r2: str, d1: str
) -> None:
context.sv_view = ScopedBackendView(
allowed_projects=frozenset({proj}),
resource_ids=frozenset({r1, r2}),
denied_resource_ids=frozenset({d1}),
)
context.sv_fragments = {}
@given('sv a fragment "{fid}" in project "{proj}" with resource "{res}"')
def step_sv_fragment_proj_res(context: Any, fid: str, proj: str, res: str) -> None:
context.sv_fragments[fid] = _make_fragment(fid, project=proj, resource=res)
@given('sv a fragment "{fid}" with no project')
def step_sv_fragment_no_proj(context: Any, fid: str) -> None:
context.sv_fragments[fid] = _make_fragment(fid, project="")
@then('sv fragment "{fid}" should be visible')
def step_sv_fragment_visible(context: Any, fid: str) -> None:
frag = context.sv_fragments[fid]
assert context.sv_view.is_visible(frag), f"Expected fragment '{fid}' to be visible"
@then('sv fragment "{fid}" should not be visible')
def step_sv_fragment_not_visible(context: Any, fid: str) -> None:
frag = context.sv_fragments[fid]
assert not context.sv_view.is_visible(frag), (
f"Expected fragment '{fid}' NOT to be visible"
)
# ---------------------------------------------------------------------------
# ScopedBackendView: backend search proxying
# ---------------------------------------------------------------------------
@given("sv a ScopedBackendView from that scope")
def step_sv_view_from_scope(context: Any) -> None:
context.sv_view = ScopedBackendView.from_resource_scope(context.sv_scope)
context.sv_fragments = {}
@given("sv an InMemoryTextBackend")
def step_sv_text_backend(context: Any) -> None:
context.sv_text_backend = InMemoryTextBackend()
@given("sv an InMemoryVectorBackend")
def step_sv_vector_backend(context: Any) -> None:
context.sv_vector_backend = InMemoryVectorBackend()
@given("sv an InMemoryGraphBackend")
def step_sv_graph_backend(context: Any) -> None:
context.sv_graph_backend = InMemoryGraphBackend()
@when('sv I search text for "{query}"')
def step_sv_search_text(context: Any, query: str) -> None:
context.sv_text_results = context.sv_view.search_text(
context.sv_text_backend, query
)
context.sv_search_error = None
@then("sv the text search should return an empty list")
def step_sv_text_empty(context: Any) -> None:
assert context.sv_text_results == []
@then("sv the text search should not raise an error")
def step_sv_text_no_error(context: Any) -> None:
assert context.sv_search_error is None
@when("sv I search vector with embedding")
def step_sv_search_vector(context: Any) -> None:
context.sv_vector_results = context.sv_view.search_vector(
context.sv_vector_backend, [0.1, 0.2, 0.3]
)
@then("sv the vector search should return an empty list")
def step_sv_vector_empty(context: Any) -> None:
assert context.sv_vector_results == []
@when('sv I search graph for "{query}"')
def step_sv_search_graph(context: Any, query: str) -> None:
context.sv_graph_result = context.sv_view.search_graph(
context.sv_graph_backend, query
)
@then("sv the graph search should return an empty GraphResult")
def step_sv_graph_empty(context: Any) -> None:
assert context.sv_graph_result.triples == []
# ---------------------------------------------------------------------------
# effective_scope
# ---------------------------------------------------------------------------
@given(
'sv a triple-resource ScopedBackendView for "{proj}" with "{r1}", "{r2}", "{r3}" denying "{d}"'
)
def step_sv_view_three_res_denied(
context: Any, proj: str, r1: str, r2: str, r3: str, d: str
) -> None:
context.sv_view = ScopedBackendView(
allowed_projects=frozenset({proj}),
resource_ids=frozenset({r1, r2, r3}),
denied_resource_ids=frozenset({d}),
)
context.sv_fragments = {}
@given('sv a single-deny ScopedBackendView for "{proj}" allow "{r}" deny "{d}"')
def step_sv_view_one_res_denied(context: Any, proj: str, r: str, d: str) -> None:
context.sv_view = ScopedBackendView(
allowed_projects=frozenset({proj}),
resource_ids=frozenset({r}),
denied_resource_ids=frozenset({d}),
)
context.sv_fragments = {}
@then('sv effective_scope should contain "{r1}" and "{r2}" but not "{r3}"')
def step_sv_effective_contains_not(context: Any, r1: str, r2: str, r3: str) -> None:
eff = context.sv_view.effective_scope()
assert r1 in eff, f"Expected {r1} in effective_scope"
assert r2 in eff, f"Expected {r2} in effective_scope"
assert r3 not in eff, f"Expected {r3} NOT in effective_scope"
@then("sv effective_scope should be empty")
def step_sv_effective_empty(context: Any) -> None:
eff = context.sv_view.effective_scope()
assert len(eff) == 0, f"Expected empty effective_scope, got {eff}"
# ---------------------------------------------------------------------------
# from_resource_scope
# ---------------------------------------------------------------------------
@when("sv I create a ScopedBackendView from that ResourceScope")
def step_sv_create_view_from_scope(context: Any) -> None:
context.sv_view = ScopedBackendView.from_resource_scope(context.sv_scope)
@then('sv the view should allow project "{proj}"')
def step_sv_view_allows_proj(context: Any, proj: str) -> None:
assert proj in context.sv_view.allowed_projects
@then('sv the view resource_ids should contain "{r1}" and "{r2}"')
def step_sv_view_res_contains(context: Any, r1: str, r2: str) -> None:
assert r1 in context.sv_view.resource_ids
assert r2 in context.sv_view.resource_ids
# ---------------------------------------------------------------------------
# ScopedBackendSet
# ---------------------------------------------------------------------------
@given("sv a ScopedBackendSet with text, vector, and graph backends")
def step_sv_backend_set_all(context: Any) -> None:
context.sv_backend_set = create_scoped_backend_set(
context.sv_scope,
text_backend=InMemoryTextBackend(),
vector_backend=InMemoryVectorBackend(),
graph_backend=InMemoryGraphBackend(),
)
@given("sv a ScopedBackendSet with no text backend")
def step_sv_backend_set_no_text(context: Any) -> None:
context.sv_backend_set = create_scoped_backend_set(
context.sv_scope,
text_backend=None,
vector_backend=InMemoryVectorBackend(),
graph_backend=InMemoryGraphBackend(),
)
@when('sv I search text via backend set for "{query}"')
def step_sv_backend_set_search(context: Any, query: str) -> None:
context.sv_backend_set_results = context.sv_backend_set.search_text(query)
@then("sv the backend set text search should succeed")
def step_sv_backend_set_ok(context: Any) -> None:
assert isinstance(context.sv_backend_set_results, list)
@then("sv the backend set text search should return empty list")
def step_sv_backend_set_empty(context: Any) -> None:
assert context.sv_backend_set_results == []
@when("sv I search vector via backend set")
def step_sv_backend_set_vector(context: Any) -> None:
context.sv_backend_set_vec_results = context.sv_backend_set.search_vector(
[0.1, 0.2, 0.3]
)
@then("sv the backend set vector search should return empty list")
def step_sv_backend_set_vec_empty(context: Any) -> None:
assert context.sv_backend_set_vec_results == []
@when('sv I search graph via backend set for "{query}"')
def step_sv_backend_set_graph(context: Any, query: str) -> None:
context.sv_backend_set_graph_result = context.sv_backend_set.search_graph(query)
@then("sv the backend set graph search should return empty GraphResult")
def step_sv_backend_set_graph_empty(context: Any) -> None:
assert context.sv_backend_set_graph_result.triples == []
# ---------------------------------------------------------------------------
# ScopeViolationError
# ---------------------------------------------------------------------------
@when('sv I raise a ScopeViolationError with resource "{res}" and projects "{proj}"')
def step_sv_raise_scope_error(context: Any, res: str, proj: str) -> None:
context.sv_error = ScopeViolationError(
"test error",
resource_ids=(res,),
scope_project_names=(proj,),
)
@then('sv the error resource_ids should contain "{res}"')
def step_sv_error_res(context: Any, res: str) -> None:
assert res in context.sv_error.resource_ids
@then('sv the error scope_project_names should contain "{proj}"')
def step_sv_error_proj(context: Any, proj: str) -> None:
assert proj in context.sv_error.scope_project_names
# ---------------------------------------------------------------------------
# validate_project_scope
# ---------------------------------------------------------------------------
@then('sv validating projects "{req}" against available "{a1}", "{a2}" should pass')
def step_sv_validate_proj_pass(context: Any, req: str, a1: str, a2: str) -> None:
validate_project_scope(frozenset({req}), frozenset({a1, a2}))
@then(
'sv validating projects "{req}" against available "{a1}", "{a2}" should raise ScopeViolationError'
)
def step_sv_validate_proj_fail_one(context: Any, req: str, a1: str, a2: str) -> None:
try:
validate_project_scope(frozenset({req}), frozenset({a1, a2}))
raise AssertionError("Expected ScopeViolationError")
except ScopeViolationError:
pass
@then(
'sv validating mixed projects "{r1}", "{r2}" against available "{a1}", "{a2}" should raise ScopeViolationError'
)
def step_sv_validate_proj_fail_two(
context: Any, r1: str, r2: str, a1: str, a2: str
) -> None:
try:
validate_project_scope(frozenset({r1, r2}), frozenset({a1, a2}))
raise AssertionError("Expected ScopeViolationError")
except ScopeViolationError:
pass
@then("sv validating empty requested projects should raise ValueError")
def step_sv_validate_proj_empty(context: Any) -> None:
try:
validate_project_scope(frozenset(), frozenset({"p1"}))
raise AssertionError("Expected ValueError")
except ValueError:
pass
# ---------------------------------------------------------------------------
# validate_resource_scope
# ---------------------------------------------------------------------------
@then('sv validating resources "{res}" against that scope should pass')
def step_sv_validate_res_pass(context: Any, res: str) -> None:
validate_resource_scope(frozenset({res}), context.sv_scope)
@then(
'sv validating resources "{res}" against that scope should raise ScopeViolationError'
)
def step_sv_validate_res_fail(context: Any, res: str) -> None:
try:
validate_resource_scope(frozenset({res}), context.sv_scope)
raise AssertionError("Expected ScopeViolationError")
except ScopeViolationError:
pass
# ---------------------------------------------------------------------------
# ResourceAliasResolver
# ---------------------------------------------------------------------------
@given('sv a project "{ns}/{name}" with resource "{res}" aliased "{alias}"')
def step_sv_proj_one_alias(
context: Any, ns: str, name: str, res: str, alias: str
) -> None:
context.sv_project = _make_project(name, namespace=ns, links=[(res, alias)])
@given(
'sv a ResourceScope with two resources "{r1}", "{r2}" and two projects "{p1}", "{p2}"'
)
def step_sv_scope_two_res_two_proj(
context: Any, r1: str, r2: str, p1: str, p2: str
) -> None:
context.sv_scope = ResourceScope(
resource_ids=frozenset({r1, r2}),
project_names=frozenset({p1, p2}),
)
@given(
'sv a project "{ns}/{name}" with two resources "{r1}" aliased "{a1}" and "{r2}" aliased "{a2}"'
)
def step_sv_proj_two_aliases(
context: Any, ns: str, name: str, r1: str, a1: str, r2: str, a2: str
) -> None:
context.sv_project = _make_project(name, namespace=ns, links=[(r1, a1), (r2, a2)])
@given('sv a project "{ns}/{name}" with duplicate alias "{alias}" on "{r1}" and "{r2}"')
def step_sv_proj_dup_alias(
context: Any, ns: str, name: str, alias: str, r1: str, r2: str
) -> None:
context.sv_project = _make_project(
name, namespace=ns, links=[(r1, alias), (r2, alias)]
)
@given("sv a ResourceAliasResolver")
def step_sv_resolver(context: Any) -> None:
context.sv_resolver = ResourceAliasResolver()
@when('sv I resolve alias "{alias}"')
def step_sv_resolve_alias(context: Any, alias: str) -> None:
context.sv_resolved = context.sv_resolver.resolve(context.sv_project, alias)
@then('sv the resolved resource_id should be "{expected}"')
def step_sv_resolved_is(context: Any, expected: str) -> None:
assert context.sv_resolved == expected, (
f"Expected '{expected}', got '{context.sv_resolved}'"
)
@then("sv the resolved resource_id should be None")
def step_sv_resolved_none(context: Any) -> None:
assert context.sv_resolved is None
@then("sv resolving empty alias should raise ValueError")
def step_sv_resolve_empty(context: Any) -> None:
try:
context.sv_resolver.resolve(context.sv_project, "")
raise AssertionError("Expected ValueError")
except ValueError:
pass
@when('sv I resolve many aliases "{a1}", "{a2}"')
def step_sv_resolve_many(context: Any, a1: str, a2: str) -> None:
context.sv_resolved_set = context.sv_resolver.resolve_many(
context.sv_project, (a1, a2)
)
@then('sv the resolved set should contain "{r1}" and "{r2}"')
def step_sv_resolved_set_contains(context: Any, r1: str, r2: str) -> None:
assert r1 in context.sv_resolved_set
assert r2 in context.sv_resolved_set
@then('sv the resolved set should contain "{r1}"')
def step_sv_resolved_set_contains_one(context: Any, r1: str) -> None:
assert r1 in context.sv_resolved_set
@when("sv I validate alias uniqueness")
def step_sv_validate_uniqueness(context: Any) -> None:
context.sv_uniqueness_errors = context.sv_resolver.validate_uniqueness(
context.sv_project
)
@then('sv there should be {count:d} uniqueness error mentioning "{alias}"')
def step_sv_uniqueness_errors_with_alias(context: Any, count: int, alias: str) -> None:
assert len(context.sv_uniqueness_errors) == count
if count > 0:
assert any(alias in err for err in context.sv_uniqueness_errors)
@then("sv there should be {count:d} uniqueness errors")
def step_sv_uniqueness_errors_count(context: Any, count: int) -> None:
assert len(context.sv_uniqueness_errors) == count
# ---------------------------------------------------------------------------
# resolve_resource_scope
# ---------------------------------------------------------------------------
@given('sv a project "{ns}/{name}" with linked resources "{r1}", "{r2}"')
def step_sv_proj_linked_resources(
context: Any, ns: str, name: str, r1: str, r2: str
) -> None:
context.sv_project = _make_project(
name, namespace=ns, links=[(r1, None), (r2, None)]
)
@when("sv I resolve the resource scope")
def step_sv_resolve_scope(context: Any) -> None:
context.sv_scope = resolve_resource_scope([context.sv_project])
@when('sv I resolve the resource scope with include "{inc}"')
def step_sv_resolve_scope_include(context: Any, inc: str) -> None:
context.sv_scope = resolve_resource_scope(
[context.sv_project], include_resources=(inc,)
)
@when('sv I resolve the resource scope with exclude "{exc}"')
def step_sv_resolve_scope_exclude(context: Any, exc: str) -> None:
context.sv_scope = resolve_resource_scope(
[context.sv_project], exclude_resources=(exc,)
)
@then("sv the scope should have {count:d} resource_ids")
def step_sv_scope_res_count(context: Any, count: int) -> None:
assert len(context.sv_scope.resource_ids) == count, (
f"Expected {count} resource_ids, got {len(context.sv_scope.resource_ids)}: "
f"{context.sv_scope.resource_ids}"
)
@then("sv resolving scope with empty projects should raise ValueError")
def step_sv_resolve_scope_empty(context: Any) -> None:
try:
resolve_resource_scope([])
raise AssertionError("Expected ValueError")
except ValueError:
pass
# ---------------------------------------------------------------------------
# Coverage: field validator coercion (scoped_view L226, L237)
# ---------------------------------------------------------------------------
@then("sv constructing view with list resource_ids should succeed")
def step_sv_view_list_res(context: Any) -> None:
view = ScopedBackendView(
allowed_projects=frozenset({"p"}),
resource_ids=["R1", "R2"], # type: ignore[arg-type]
)
assert "R1" in view.resource_ids
assert isinstance(view.resource_ids, frozenset)
@then("sv constructing view with list denied_resource_ids should succeed")
def step_sv_view_list_denied(context: Any) -> None:
view = ScopedBackendView(
allowed_projects=frozenset({"p"}),
resource_ids=frozenset({"R1", "R2"}),
denied_resource_ids=["R2"], # type: ignore[arg-type]
)
assert "R2" in view.denied_resource_ids
assert isinstance(view.denied_resource_ids, frozenset)
# ---------------------------------------------------------------------------
# Coverage: _require_effective_scope raises (scoped_view L326-330)
# ---------------------------------------------------------------------------
@then("sv searching text via denied view should raise ScopeViolationError")
def step_sv_search_text_denied(context: Any) -> None:
try:
context.sv_view.search_text(context.sv_text_backend, "test")
raise AssertionError("Expected ScopeViolationError")
except ScopeViolationError:
pass
# ---------------------------------------------------------------------------
# Coverage: ScopedBackendSet None backends (scoped_view L489, L505)
# ---------------------------------------------------------------------------
@given("sv a ScopedBackendSet with no vector backend")
def step_sv_backend_set_no_vector(context: Any) -> None:
context.sv_backend_set_novector = create_scoped_backend_set(
context.sv_scope,
text_backend=InMemoryTextBackend(),
vector_backend=None,
graph_backend=InMemoryGraphBackend(),
)
@then("sv searching vector via no-vector set should return empty list")
def step_sv_novector_search(context: Any) -> None:
result = context.sv_backend_set_novector.search_vector([0.1])
assert result == []
@given("sv a ScopedBackendSet with no graph backend")
def step_sv_backend_set_no_graph(context: Any) -> None:
context.sv_backend_set_nograph = create_scoped_backend_set(
context.sv_scope,
text_backend=InMemoryTextBackend(),
vector_backend=InMemoryVectorBackend(),
graph_backend=None,
)
@then("sv searching graph via no-graph set should return empty GraphResult")
def step_sv_nograph_search(context: Any) -> None:
result = context.sv_backend_set_nograph.search_graph("SELECT ?s")
assert result.triples == []
# ---------------------------------------------------------------------------
# Coverage: resolve_many unresolvable (scoped_view L617)
# ---------------------------------------------------------------------------
@then("sv the resolved set size should be {count:d}")
def step_sv_resolved_set_size(context: Any, count: int) -> None:
assert len(context.sv_resolved_set) == count, (
f"Expected {count} entries, got {len(context.sv_resolved_set)}"
)
# ---------------------------------------------------------------------------
# Coverage: validate_uniqueness with no-alias resource (scoped_view L638)
# ---------------------------------------------------------------------------
@given("sv a project with aliased and unaliased resources")
def step_sv_proj_mixed_alias(context: Any) -> None:
context.sv_project = _make_project(
"myproj",
namespace="local",
links=[
("01HQ8ZDRX50000000000000001", "repo"),
("01HQ8ZDRX50000000000000002", None),
],
)
# ---------------------------------------------------------------------------
# Coverage: validate_project_scope empty available
# ---------------------------------------------------------------------------
@then("sv validating projects against empty available should raise ValueError")
def step_sv_validate_proj_empty_avail(context: Any) -> None:
try:
validate_project_scope(frozenset({"p1"}), frozenset())
raise AssertionError("Expected ValueError")
except ValueError:
pass
# ---------------------------------------------------------------------------
# Coverage: validate_resource_scope empty requested
# ---------------------------------------------------------------------------
@then("sv validating empty resources against scope should pass")
def step_sv_validate_res_empty_req(context: Any) -> None:
validate_resource_scope(frozenset(), context.sv_scope)
# ---------------------------------------------------------------------------
# P0-S1: is_visible empty resource_id deny-by-default
# ---------------------------------------------------------------------------
@given('sv a ScopedBackendView for projects "{proj}" with resource_ids "{rid}"')
def step_sv_view_with_resources(context: Any, proj: str, rid: str) -> None:
context.sv_view = ScopedBackendView(
allowed_projects=frozenset([proj]),
resource_ids=frozenset([rid]),
)
@given('sv a ScopedBackendView for project "{proj}"')
def step_sv_view_project_only(context: Any, proj: str) -> None:
context.sv_view = ScopedBackendView(
allowed_projects=frozenset([proj]),
)
@when('sv I check visibility of a fragment with project "{proj}" and empty resource_id')
def step_sv_check_empty_rid(context: Any, proj: str) -> None:
frag = TieredFragment(
fragment_id="empty-rid-frag",
content="test",
project_name=proj,
resource_id="",
tier=ContextTier.HOT,
)
context.sv_vis_result = context.sv_view.is_visible(frag)
@then("sv the fragment should NOT be visible")
def step_sv_not_visible(context: Any) -> None:
assert context.sv_vis_result is False
@then("sv the fragment should be visible")
def step_sv_is_visible(context: Any) -> None:
assert context.sv_vis_result is True
# ---------------------------------------------------------------------------
# P2-B2: tuple coercion
# ---------------------------------------------------------------------------
@when("sv I create a ScopedBackendView with tuple allowed_projects")
def step_sv_tuple_projects(context: Any) -> None:
context.sv_view = ScopedBackendView(
allowed_projects=("proj-a",),
resource_ids=frozenset(),
)
@then("sv the view should have {count:d} allowed project")
def step_sv_view_project_count(context: Any, count: int) -> None:
assert len(context.sv_view.allowed_projects) == count
+282
View File
@@ -0,0 +1,282 @@
"""Step definitions for ContextTierService scope enforcement tests.
Extracted from ``scoped_view_steps.py`` to keep step files manageable.
All steps prefixed with ``sv`` to avoid AmbiguousStep collisions.
"""
from __future__ import annotations
from typing import Any
from behave import given, then, when # type: ignore[import-untyped]
from cleveragents.domain.models.acms.scope_resolution import (
resolve_resource_scope,
)
from cleveragents.domain.models.acms.scoped_view import (
ScopeViolationError,
)
from cleveragents.domain.models.acms.tiers import (
ContextTier,
TieredFragment,
)
__all__: list[str] = []
def _make_fragment(
fid: str,
project: str = "",
resource: str = "",
tier: str = "hot",
) -> TieredFragment:
"""Create a TieredFragment for testing."""
return TieredFragment(
fragment_id=fid,
content=f"content-{fid}",
tier=ContextTier(tier),
project_name=project,
resource_id=resource,
)
# ---------------------------------------------------------------------------
# ContextTierService: scope enforcement
# ---------------------------------------------------------------------------
@given("sv a ContextTierService")
def step_sv_tier_service(context: Any) -> None:
from cleveragents.application.services.context_tiers import ContextTierService
context.sv_tier_svc = ContextTierService()
@when('sv I store fragment "{fid}" project "{proj}" resource "{res}" tier "{tier}"')
def step_sv_store_fragment(
context: Any, fid: str, proj: str, res: str, tier: str
) -> None:
frag = _make_fragment(fid, project=proj, resource=res, tier=tier)
context.sv_tier_svc.store(frag)
@then("sv get_scoped_by_resource should return {count:d} fragment")
def step_sv_scoped_by_resource(context: Any, count: int) -> None:
result = context.sv_tier_svc.get_scoped_by_resource(context.sv_scope)
assert len(result) == count, f"Expected {count} fragments, got {len(result)}"
@then('sv storing fragment with resource "{res}" should raise ScopeViolationError')
def step_sv_store_scope_check(context: Any, res: str) -> None:
frag = _make_fragment("bad-frag", project="proj-a", resource=res)
try:
context.sv_tier_svc.store_with_scope_check(frag, context.sv_scope)
raise AssertionError("Expected ScopeViolationError")
except ScopeViolationError:
pass
@then(
'sv storing fragment with project "{proj}" and resource "{res}"'
" should raise ScopeViolationError"
)
def step_sv_store_scope_check_project(context: Any, proj: str, res: str) -> None:
frag = _make_fragment("bad-proj-frag", project=proj, resource=res)
try:
context.sv_tier_svc.store_with_scope_check(frag, context.sv_scope)
raise AssertionError("Expected ScopeViolationError for out-of-scope project")
except ScopeViolationError:
pass
@then('sv validating fragment "{fid}" against the scope should pass')
def step_sv_validate_fragment(context: Any, fid: str) -> None:
context.sv_tier_svc.validate_fragment_scope(fid, context.sv_scope)
@then(
'sv validating fragment "{fid}" against the scope should raise ScopeViolationError'
)
def step_sv_validate_fragment_fail(context: Any, fid: str) -> None:
try:
context.sv_tier_svc.validate_fragment_scope(fid, context.sv_scope)
raise AssertionError("Expected ScopeViolationError")
except ScopeViolationError:
pass
# ---------------------------------------------------------------------------
# Coverage: context_tiers store empty id
# ---------------------------------------------------------------------------
@then("sv storing fragment with empty id should raise ValueError")
def step_sv_store_empty_id(context: Any) -> None:
frag = TieredFragment.model_construct(
fragment_id="",
content="content",
tier=ContextTier.HOT,
project_name="p",
resource_id="R1",
)
try:
context.sv_tier_svc.store(frag)
raise AssertionError("Expected ValueError")
except ValueError:
pass
# ---------------------------------------------------------------------------
# Coverage: context_tiers get_scoped_view empty
# ---------------------------------------------------------------------------
@then("sv get_scoped_view with empty projects should return empty list")
def step_sv_scoped_view_empty(context: Any) -> None:
result = context.sv_tier_svc.get_scoped_view([])
assert result == []
# ---------------------------------------------------------------------------
# Coverage: context_tiers evict_lru empty
# ---------------------------------------------------------------------------
@then("sv evict_lru on empty cold tier should return empty list")
def step_sv_evict_empty(context: Any) -> None:
result = context.sv_tier_svc.evict_lru(ContextTier.COLD, 1)
assert result == []
# ---------------------------------------------------------------------------
# Coverage: context_tiers get_scoped_metrics
# ---------------------------------------------------------------------------
@then('sv get_scoped_metrics for "{proj}" should show {h:d} hot and {w:d} warm')
def step_sv_scoped_metrics(context: Any, proj: str, h: int, w: int) -> None:
metrics = context.sv_tier_svc.get_scoped_metrics([proj])
assert metrics.hot_count == h, f"Expected hot={h}, got {metrics.hot_count}"
assert metrics.warm_count == w, f"Expected warm={w}, got {metrics.warm_count}"
# ---------------------------------------------------------------------------
# Coverage: context_tiers validate_fragment_scope not found
# ---------------------------------------------------------------------------
@then("sv validating non-existent fragment should raise ValueError")
def step_sv_validate_nonexistent(context: Any) -> None:
try:
context.sv_tier_svc.validate_fragment_scope("ghost", context.sv_scope)
raise AssertionError("Expected ValueError")
except ValueError:
pass
# ---------------------------------------------------------------------------
# Coverage: context_tiers store_with_scope_check happy path
# ---------------------------------------------------------------------------
@then("sv store_with_scope_check should accept in-scope fragment")
def step_sv_store_scope_ok(context: Any) -> None:
frag = _make_fragment("ok-frag", project="proj-a", resource="RES01")
context.sv_tier_svc.store_with_scope_check(frag, context.sv_scope)
result = context.sv_tier_svc.get("ok-frag")
assert result is not None
# ---------------------------------------------------------------------------
# Coverage: context_tiers budget property
# ---------------------------------------------------------------------------
@then("sv tier service budget should be a TierBudget")
def step_sv_budget(context: Any) -> None:
from cleveragents.domain.models.acms.tiers import TierBudget
budget = context.sv_tier_svc.budget
assert isinstance(budget, TierBudget)
# ---------------------------------------------------------------------------
# Coverage: context_tiers get_for_actor no project filter
# ---------------------------------------------------------------------------
@then("sv get_for_actor without projects should return a list")
def step_sv_actor_no_projects(context: Any) -> None:
from cleveragents.domain.models.acms.tiers import ActorRole
result = context.sv_tier_svc.get_for_actor(ActorRole.STRATEGIST)
assert isinstance(result, list)
# ---------------------------------------------------------------------------
# Coverage: context_tiers _summarize_for_cold false branch
# ---------------------------------------------------------------------------
@when('sv I store a short warm fragment "{fid}"')
def step_sv_store_short_warm(context: Any, fid: str) -> None:
frag = TieredFragment(
fragment_id=fid,
content="short",
tier=ContextTier.WARM,
project_name="proj-a",
resource_id="RES01",
)
context.sv_tier_svc.store(frag)
@then('sv demoting "{fid}" to cold should keep content intact')
def step_sv_demote_short(context: Any, fid: str) -> None:
result = context.sv_tier_svc.demote(fid)
assert result is not None
assert result.content == "short"
assert result.tier == ContextTier.COLD
# ---------------------------------------------------------------------------
# P0-S2: get_scoped
# ---------------------------------------------------------------------------
@then('sv get_scoped for "{fid}" should return the fragment')
def step_sv_get_scoped_found(context: Any, fid: str) -> None:
result = context.sv_tier_svc.get_scoped(fid, context.sv_scope)
assert result is not None
assert result.fragment_id == fid
@then('sv get_scoped for "{fid}" should return None')
def step_sv_get_scoped_none(context: Any, fid: str) -> None:
result = context.sv_tier_svc.get_scoped(fid, context.sv_scope)
assert result is None
# ---------------------------------------------------------------------------
# P1-SPEC1: DAG expansion
# ---------------------------------------------------------------------------
@when('sv I resolve scope with registry_lookup expanding "{parent}" to "{child}"')
def step_sv_resolve_with_dag(context: Any, parent: str, child: str) -> None:
def lookup(rid: str) -> frozenset[str]:
if rid == parent:
return frozenset([child])
return frozenset()
context.sv_scope = resolve_resource_scope(
[context.sv_project],
registry_lookup=lookup,
)
@when(
'sv I store fragment "{fid}" with project "{proj}" and empty resource_id in tier "{tier}"'
)
def step_sv_store_empty_rid(context: Any, fid: str, proj: str, tier: str) -> None:
frag = _make_fragment(fid, project=proj, resource="", tier=tier)
context.sv_tier_svc.store(frag)
+273
View File
@@ -0,0 +1,273 @@
"""Robot Framework helper for scoped backend view smoke tests.
Provides a CLI-style interface for Robot to invoke scoped view creation,
resource scope resolution, backend proxying, and scope validation.
Exit code 0 = success, 1 = failure.
Usage:
python robot/helper_scoped_view.py <command>
"""
from __future__ import annotations
import sys
from pathlib import Path
# Ensure the src directory is on the import path.
_SRC = str(Path(__file__).resolve().parents[1] / "src")
if _SRC not in sys.path:
sys.path.insert(0, _SRC)
from cleveragents.domain.models.acms.scope_resolution import ( # noqa: E402
ResourceAliasResolver,
resolve_resource_scope,
validate_project_scope,
validate_resource_scope,
)
from cleveragents.domain.models.acms.scoped_view import ( # noqa: E402
ResourceScope,
ScopedBackendView,
ScopeViolationError,
create_scoped_backend_set,
)
from cleveragents.domain.models.acms.stubs import ( # noqa: E402
InMemoryGraphBackend,
InMemoryTextBackend,
InMemoryVectorBackend,
)
from cleveragents.domain.models.acms.tiers import ( # noqa: E402
ContextTier,
TieredFragment,
)
from cleveragents.domain.models.core.project import ( # noqa: E402
LinkedResource,
NamespacedProject,
)
def main() -> int:
"""Entry point called by Robot Framework ``Run Process``."""
if len(sys.argv) < 2:
print("Usage: helper_scoped_view.py <command>")
return 1
command: str = sys.argv[1]
if command == "resource-scope":
try:
scope = ResourceScope(
resource_ids=frozenset({"R1", "R2"}),
project_names=frozenset({"local/api"}),
)
assert scope.contains_resource("R1")
assert scope.contains_resource("R2")
assert not scope.contains_resource("R99")
assert scope.contains_project("local/api")
assert not scope.contains_project("local/frontend")
assert scope.temporal_scope == "current"
print("scoped-view-resource-scope-ok")
return 0
except Exception as exc:
print(f"scoped-view-resource-scope-fail: {exc}")
return 1
if command == "path-filtering":
try:
scope = ResourceScope(
resource_ids=frozenset({"R1"}),
project_names=frozenset({"p1"}),
include_paths=("src/**/*.py",),
exclude_paths=("**/test_*",),
)
assert scope.matches_path("src/main.py")
assert not scope.matches_path("src/test_main.py")
assert not scope.matches_path("docs/readme.md")
print("scoped-view-path-filtering-ok")
return 0
except Exception as exc:
print(f"scoped-view-path-filtering-fail: {exc}")
return 1
if command == "fragment-filtering":
try:
view = ScopedBackendView(
allowed_projects=frozenset({"proj-a"}),
resource_ids=frozenset({"R1"}),
)
frag_ok = TieredFragment(
fragment_id="f1",
content="c",
tier=ContextTier("hot"),
project_name="proj-a",
resource_id="R1",
)
frag_bad_proj = TieredFragment(
fragment_id="f2",
content="c",
tier=ContextTier("hot"),
project_name="proj-b",
resource_id="R1",
)
frag_bad_res = TieredFragment(
fragment_id="f3",
content="c",
tier=ContextTier("hot"),
project_name="proj-a",
resource_id="R99",
)
assert view.is_visible(frag_ok)
assert not view.is_visible(frag_bad_proj)
assert not view.is_visible(frag_bad_res)
print("scoped-view-fragment-filtering-ok")
return 0
except Exception as exc:
print(f"scoped-view-fragment-filtering-fail: {exc}")
return 1
if command == "backend-proxying":
try:
scope = ResourceScope(
resource_ids=frozenset({"R1"}),
project_names=frozenset({"local/api"}),
)
view = ScopedBackendView.from_resource_scope(scope)
text_results = view.search_text(InMemoryTextBackend(), "test")
assert text_results == []
vec_results = view.search_vector(InMemoryVectorBackend(), [0.1, 0.2, 0.3])
assert vec_results == []
graph_result = view.search_graph(
InMemoryGraphBackend(), "SELECT ?s WHERE { ?s a uko:X }"
)
assert graph_result.triples == []
print("scoped-view-backend-proxying-ok")
return 0
except Exception as exc:
print(f"scoped-view-backend-proxying-fail: {exc}")
return 1
if command == "backend-set":
try:
scope = ResourceScope(
resource_ids=frozenset({"R1"}),
project_names=frozenset({"local/api"}),
)
bset = create_scoped_backend_set(
scope,
text_backend=InMemoryTextBackend(),
vector_backend=InMemoryVectorBackend(),
graph_backend=InMemoryGraphBackend(),
)
assert bset.search_text("q") == []
assert bset.search_vector([0.1]) == []
assert bset.search_graph("SELECT ?s WHERE { ?s a uko:X }").triples == []
# None backend returns empty
bset_no_text = create_scoped_backend_set(scope)
assert bset_no_text.search_text("q") == []
print("scoped-view-backend-set-ok")
return 0
except Exception as exc:
print(f"scoped-view-backend-set-fail: {exc}")
return 1
if command == "scope-validation":
try:
# validate_project_scope: valid
validate_project_scope(
frozenset({"proj-a"}), frozenset({"proj-a", "proj-b"})
)
# validate_project_scope: invalid
raised = False
try:
validate_project_scope(
frozenset({"proj-c"}), frozenset({"proj-a", "proj-b"})
)
except ScopeViolationError:
raised = True
assert raised, "Expected ScopeViolationError for out-of-scope project"
# validate_resource_scope: valid
scope = ResourceScope(
resource_ids=frozenset({"R1", "R2"}),
project_names=frozenset({"p1"}),
)
validate_resource_scope(frozenset({"R1"}), scope)
# validate_resource_scope: invalid
raised = False
try:
validate_resource_scope(frozenset({"R99"}), scope)
except ScopeViolationError:
raised = True
assert raised, "Expected ScopeViolationError for out-of-scope resource"
print("scoped-view-scope-validation-ok")
return 0
except Exception as exc:
print(f"scoped-view-scope-validation-fail: {exc}")
return 1
if command == "alias-resolver":
try:
ulid1 = "01HQ8ZDRX50000000000000001"
ulid2 = "01HQ8ZDRX50000000000000002"
proj = NamespacedProject(
name="myproj",
namespace="local",
linked_resources=[
LinkedResource(resource_id=ulid1, alias="repo"),
LinkedResource(resource_id=ulid2, alias="docs"),
],
)
resolver = ResourceAliasResolver()
# Resolve by alias
assert resolver.resolve(proj, "repo") == ulid1
# Resolve by ID
assert resolver.resolve(proj, ulid1) == ulid1
# Unknown returns None
assert resolver.resolve(proj, "nonexistent") is None
# resolve_many
resolved = resolver.resolve_many(proj, ("repo", "docs"))
assert ulid1 in resolved
assert ulid2 in resolved
# validate_uniqueness: no dupes
assert resolver.validate_uniqueness(proj) == []
print("scoped-view-alias-resolver-ok")
return 0
except Exception as exc:
print(f"scoped-view-alias-resolver-fail: {exc}")
return 1
if command == "resolve-scope":
try:
ulid1 = "01HQ8ZDRX50000000000000001"
ulid2 = "01HQ8ZDRX50000000000000002"
proj = NamespacedProject(
name="api",
namespace="local",
linked_resources=[
LinkedResource(resource_id=ulid1, alias="repo"),
LinkedResource(resource_id=ulid2, alias="docs"),
],
)
# Full scope
scope = resolve_resource_scope([proj])
assert len(scope.resource_ids) == 2
# Include filter
scope_inc = resolve_resource_scope([proj], include_resources=("repo",))
assert len(scope_inc.resource_ids) == 1
assert ulid1 in scope_inc.resource_ids
# Exclude filter
scope_exc = resolve_resource_scope([proj], exclude_resources=("docs",))
assert len(scope_exc.resource_ids) == 1
assert ulid1 in scope_exc.resource_ids
print("scoped-view-resolve-scope-ok")
return 0
except Exception as exc:
print(f"scoped-view-resolve-scope-fail: {exc}")
return 1
print(f"Unknown command: {command}")
return 1
if __name__ == "__main__":
sys.exit(main())
+73
View File
@@ -0,0 +1,73 @@
*** Settings ***
Documentation Smoke tests for ACMS Scoped Backend View filtering
Resource ${CURDIR}/common.resource
Suite Setup Setup Test Environment
Suite Teardown Cleanup Test Environment
*** Variables ***
${HELPER} ${CURDIR}/helper_scoped_view.py
*** Test Cases ***
ResourceScope Creation And Containment
[Documentation] Verify ResourceScope creation, containment checks, and temporal default
${result}= Run Process ${PYTHON} ${HELPER} resource-scope cwd=${WORKSPACE}
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} scoped-view-resource-scope-ok
ResourceScope Path Filtering
[Documentation] Verify include/exclude path glob matching with recursive patterns
${result}= Run Process ${PYTHON} ${HELPER} path-filtering cwd=${WORKSPACE}
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} scoped-view-path-filtering-ok
ScopedBackendView Fragment Filtering
[Documentation] Verify fragment visibility filtering by project and resource
${result}= Run Process ${PYTHON} ${HELPER} fragment-filtering cwd=${WORKSPACE}
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} scoped-view-fragment-filtering-ok
ScopedBackendView Backend Proxying
[Documentation] Verify text/vector/graph search proxying with scope injection
${result}= Run Process ${PYTHON} ${HELPER} backend-proxying cwd=${WORKSPACE}
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} scoped-view-backend-proxying-ok
ScopedBackendSet Integration
[Documentation] Verify ScopedBackendSet wraps multiple backends with scope enforcement
${result}= Run Process ${PYTHON} ${HELPER} backend-set cwd=${WORKSPACE}
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} scoped-view-backend-set-ok
Scope Validation Guards
[Documentation] Verify validate_project_scope and validate_resource_scope raise correctly
${result}= Run Process ${PYTHON} ${HELPER} scope-validation cwd=${WORKSPACE}
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} scoped-view-scope-validation-ok
ResourceAliasResolver
[Documentation] Verify alias resolution, resolve_many, and uniqueness validation
${result}= Run Process ${PYTHON} ${HELPER} alias-resolver cwd=${WORKSPACE}
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} scoped-view-alias-resolver-ok
Resolve Resource Scope With Filters
[Documentation] Verify resolve_resource_scope with include/exclude allowlist/denylist
${result}= Run Process ${PYTHON} ${HELPER} resolve-scope cwd=${WORKSPACE}
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} scoped-view-resolve-scope-ok
@@ -1,34 +1,35 @@
"""Context Tier Service for hot/warm/cold fragment management.
The ``ContextTierService`` manages the lifecycle of context fragments
across three storage tiers:
Core tier lifecycle: store, get, promote, demote, evict, metrics.
Scope enforcement methods live in ``scoped_tiers.py`` and are mixed
in via ``ScopedTierMixin``.
- **Hot** — in-memory ``dict``, fastest access, token-budget limited.
- **Warm** — SQLite-backed (stub), moderate latency, decision-count limited.
- **Cold** — file-backed (stub), highest latency, summarised on demotion.
Supports promotion (cold→warm→hot), demotion (hot→warm→cold) with a
summarisation hook, LRU eviction, per-actor filtered views, and
project-scoped isolation via ``ScopedBackendView``.
Based on ``docs/specification.md`` ACMS tier sections and issue #208.
Based on ``docs/specification.md`` ACMS tier sections, issue #208,
and issue #193 (scoped backend view filtering).
"""
from __future__ import annotations
from datetime import UTC, datetime
import structlog
from cleveragents.application.services.scoped_tiers import ScopedTierMixin
from cleveragents.config.settings import Settings
from cleveragents.domain.models.acms.scoped_view import (
ScopedBackendView,
)
from cleveragents.domain.models.acms.tiers import (
ActorContextView,
ActorRole,
ContextTier,
ScopedBackendView,
TierBudget,
TieredFragment,
TierMetrics,
)
logger = structlog.get_logger(__name__)
# ---------------------------------------------------------------------------
# Default budget when settings are not provided
# ---------------------------------------------------------------------------
@@ -41,12 +42,12 @@ _DEFAULT_MAX_DECISIONS_COLD = 5000
_COLD_SUMMARY_MAX_CHARS = 200
class ContextTierService:
class ContextTierService(ScopedTierMixin):
"""Manage context fragments across hot/warm/cold tiers.
The service is designed for dependency injection via the DI
container. All mutable state lives in the three tier stores
and the metrics counters.
Scope enforcement (``get_scoped``, ``get_scoped_by_resource``,
``validate_fragment_scope``, ``store_with_scope_check``,
``get_scoped_metrics``) is provided by ``ScopedTierMixin``.
"""
def __init__(self, settings: Settings | None = None) -> None:
@@ -146,8 +147,16 @@ class ContextTierService:
# Apply project scoping
if project_names:
scoped = ScopedBackendView(allowed_projects=project_names)
scoped = ScopedBackendView(
allowed_projects=frozenset(project_names),
)
candidates = [f for f in candidates if scoped.is_visible(f)]
logger.debug(
"tier.actor_view_filtered",
actor_role=actor_role.value,
after=len(candidates),
project_names=project_names,
)
# Sort by last_accessed descending, cap at max_fragments
candidates.sort(key=lambda f: f.last_accessed, reverse=True)
@@ -168,11 +177,18 @@ class ContextTierService:
"""
if not project_names:
return []
scoped = ScopedBackendView(allowed_projects=project_names)
scoped = ScopedBackendView(
allowed_projects=frozenset(project_names),
)
all_frags: list[TieredFragment] = []
for store in (self._hot, self._warm, self._cold):
all_frags.extend(f for f in store.values() if scoped.is_visible(f))
all_frags.sort(key=lambda f: f.last_accessed, reverse=True)
logger.debug(
"tier.scoped_view_filtered",
project_names=project_names,
result_count=len(all_frags),
)
return all_frags
# ------------------------------------------------------------------
@@ -266,32 +282,9 @@ class ContextTierService:
warm_miss_count=self._warm_miss,
)
def get_scoped_metrics(self, project_names: list[str]) -> TierMetrics:
"""Return metrics scoped to the given project names.
Fragment counts (``hot_count``, ``warm_count``, ``cold_count``)
reflect only fragments belonging to the specified projects.
Hit/miss counters remain global as they are service-level cache
performance metrics that are not tracked per-project.
Args:
project_names: Project names to scope fragment counts to.
Raises:
ValueError: If *project_names* is empty.
"""
if not project_names:
raise ValueError("project_names must be non-empty")
scoped = ScopedBackendView(allowed_projects=project_names)
return TierMetrics(
hot_count=sum(1 for f in self._hot.values() if scoped.is_visible(f)),
warm_count=sum(1 for f in self._warm.values() if scoped.is_visible(f)),
cold_count=sum(1 for f in self._cold.values() if scoped.is_visible(f)),
hot_hit_count=self._hot_hit,
hot_miss_count=self._hot_miss,
warm_hit_count=self._warm_hit,
warm_miss_count=self._warm_miss,
)
# Scope enforcement methods (get_scoped, get_scoped_by_resource,
# validate_fragment_scope, store_with_scope_check, get_scoped_metrics)
# are provided by ScopedTierMixin — see scoped_tiers.py.
# ------------------------------------------------------------------
# Budget
@@ -306,6 +299,19 @@ class ContextTierService:
# Internal helpers
# ------------------------------------------------------------------
def _find_fragment(self, fragment_id: str) -> TieredFragment | None:
"""Look up a fragment across all tiers **without** side effects.
Unlike :meth:`get`, this does not call :meth:`_touch` and
therefore leaves ``last_accessed``, ``access_count``, and
hit/miss counters unchanged. Used by validation methods that
must be read-only.
"""
for store in (self._hot, self._warm, self._cold):
if fragment_id in store:
return store[fragment_id]
return None
def _store_for_tier(
self,
tier: ContextTier,
@@ -0,0 +1,219 @@
"""Scope enforcement mixin for the Context Tier Service.
Extracted from ``context_tiers.py`` to keep modules under the 500-line
guideline. Provides scope-aware retrieval, storage, validation, and
metrics methods that are mixed into ``ContextTierService``.
All methods assume access to ``_hot``, ``_warm``, ``_cold`` tier stores
and the ``_find_fragment``, ``store``, ``get`` helpers defined on the
host class.
Based on ``docs/specification.md`` ACMS scoped views (lines 42501--42566)
and Forgejo issue #193.
"""
from __future__ import annotations
import structlog
from cleveragents.domain.models.acms.scope_resolution import (
validate_resource_scope,
)
from cleveragents.domain.models.acms.scoped_view import (
ResourceScope,
ScopedBackendView,
ScopeViolationError,
)
from cleveragents.domain.models.acms.tiers import (
TieredFragment,
TierMetrics,
)
logger = structlog.get_logger(__name__)
class ScopedTierMixin:
"""Mixin providing scope enforcement for ``ContextTierService``.
Requires the host class to provide:
- ``_hot``, ``_warm``, ``_cold``: ``dict[str, TieredFragment]``
- ``_hot_hit``, ``_hot_miss``, ``_warm_hit``, ``_warm_miss``: int
- ``get(fragment_id: str) -> TieredFragment | None``
- ``store(fragment: TieredFragment) -> None``
- ``_find_fragment(fragment_id: str) -> TieredFragment | None``
"""
# Declared for type checkers; actual attrs live on the host class.
_hot: dict[str, TieredFragment]
_warm: dict[str, TieredFragment]
_cold: dict[str, TieredFragment]
_hot_hit: int
_hot_miss: int
_warm_hit: int
_warm_miss: int
def get(self, fragment_id: str) -> TieredFragment | None: ...
def store(self, fragment: TieredFragment) -> None: ...
def _find_fragment(self, fragment_id: str) -> TieredFragment | None: ...
# ------------------------------------------------------------------
# Scope-aware retrieval
# ------------------------------------------------------------------
def get_scoped(
self,
fragment_id: str,
scope: ResourceScope,
) -> TieredFragment | None:
"""Retrieve a fragment by ID with scope enforcement.
Unlike :meth:`get`, this validates the fragment against the
provided scope before returning it. Returns ``None`` and
logs a warning for out-of-scope fragments.
Args:
fragment_id: The fragment to retrieve.
scope: The active resource scope to enforce.
"""
frag = self.get(fragment_id)
if frag is None:
return None
scoped = ScopedBackendView.from_resource_scope(scope)
if not scoped.is_visible(frag):
logger.warning(
"tier.get_scoped_denied",
fragment_id=fragment_id,
fragment_project=frag.project_name,
fragment_resource=frag.resource_id,
scope_projects=sorted(scope.project_names),
)
return None
return frag
def get_scoped_by_resource(
self,
scope: ResourceScope,
) -> list[TieredFragment]:
"""Return fragments filtered by a full ``ResourceScope``.
Applies both project-name and resource-ID filtering.
Args:
scope: The resolved resource scope for the plan.
"""
scoped = ScopedBackendView.from_resource_scope(scope)
all_frags: list[TieredFragment] = []
rejected_count = 0
for store in (self._hot, self._warm, self._cold):
for frag in store.values():
if scoped.is_visible(frag):
all_frags.append(frag)
else:
rejected_count += 1
all_frags.sort(key=lambda f: f.last_accessed, reverse=True)
if rejected_count:
logger.info(
"tier.scope_enforcement",
accepted=len(all_frags),
rejected=rejected_count,
project_names=sorted(scope.project_names),
resource_count=len(scope.resource_ids),
)
return all_frags
def validate_fragment_scope(
self,
fragment_id: str,
scope: ResourceScope,
) -> None:
"""Validate that a specific fragment is within scope.
Raises:
ScopeViolationError: If the fragment is out of scope.
ValueError: If the fragment does not exist.
"""
frag = self._find_fragment(fragment_id)
if frag is None:
raise ValueError(f"Fragment not found: {fragment_id}")
scoped = ScopedBackendView.from_resource_scope(scope)
if not scoped.is_visible(frag):
logger.warning(
"tier.fragment_scope_violation",
fragment_id=fragment_id,
fragment_project=frag.project_name,
fragment_resource=frag.resource_id,
scope_projects=sorted(scope.project_names),
)
raise ScopeViolationError(
f"Fragment '{fragment_id}' is not within the active scope",
resource_ids=(frag.resource_id,) if frag.resource_id else (),
scope_project_names=tuple(sorted(scope.project_names)),
)
def store_with_scope_check(
self,
fragment: TieredFragment,
scope: ResourceScope,
) -> None:
"""Store a fragment after validating it is within scope.
Checks both the fragment's project name and resource ID
against the active scope before storing.
.. note::
Fragments with an empty ``project_name`` bypass the
project check (there is no project to enforce against).
Such fragments are still invisible to scoped reads
because ``ScopedBackendView.is_visible()`` rejects
empty project names.
Raises:
ScopeViolationError: If the fragment references an
out-of-scope project or resource.
"""
if fragment.project_name and not scope.contains_project(fragment.project_name):
logger.warning(
"tier.store_project_violation",
fragment_id=fragment.fragment_id,
fragment_project=fragment.project_name,
scope_projects=sorted(scope.project_names),
)
raise ScopeViolationError(
f"Fragment project '{fragment.project_name}' "
f"is not within the active scope",
scope_project_names=tuple(sorted(scope.project_names)),
)
if fragment.resource_id:
validate_resource_scope(
frozenset({fragment.resource_id}),
scope,
)
self.store(fragment)
def get_scoped_metrics(self, project_names: list[str]) -> TierMetrics:
"""Return metrics scoped to the given project names.
Fragment counts reflect only fragments belonging to the
specified projects. Hit/miss counters remain global.
Raises:
ValueError: If *project_names* is empty.
"""
if not project_names:
raise ValueError("project_names must be non-empty")
scoped = ScopedBackendView(
allowed_projects=frozenset(project_names),
)
return TierMetrics(
hot_count=sum(1 for f in self._hot.values() if scoped.is_visible(f)),
warm_count=sum(1 for f in self._warm.values() if scoped.is_visible(f)),
cold_count=sum(1 for f in self._cold.values() if scoped.is_visible(f)),
hot_hit_count=self._hot_hit,
hot_miss_count=self._hot_miss,
warm_hit_count=self._warm_hit,
warm_miss_count=self._warm_miss,
)
@@ -46,7 +46,13 @@ Tier types (from :mod:`~cleveragents.domain.models.acms.tiers`):
- ``TierBudget`` -- Per-tier token/decision budget limits
- ``ActorContextView`` -- Per-actor filtered view configuration
- ``TierMetrics`` -- Hit/miss counters for cache monitoring
- ``ScopedBackendView`` -- Project-scoped resource isolation
Scoped view types (from :mod:`~cleveragents.domain.models.acms.scoped_view`):
- ``ResourceScope`` -- Resolved resource/project scope for a plan
- ``ScopeViolationError`` -- Raised on out-of-scope access attempts
- ``ScopedBackendView`` -- Project-scoped backend query filtering
- ``ScopedBackendSet`` -- Bundle of scoped text/vector/graph views
- ``ResourceAliasResolver`` -- Alias-to-resource-ID resolution helper
Analyzer types (from :mod:`~cleveragents.domain.models.acms.analyzers`):
- ``UKOTriple`` -- Immutable subject-predicate-object triple
@@ -85,6 +91,19 @@ from cleveragents.domain.models.acms.crp import (
)
from cleveragents.domain.models.acms.markdown_analyzer import MarkdownAnalyzer
from cleveragents.domain.models.acms.python_analyzer import PythonAnalyzer
from cleveragents.domain.models.acms.scope_resolution import (
ResourceAliasResolver,
resolve_resource_scope,
validate_project_scope,
validate_resource_scope,
)
from cleveragents.domain.models.acms.scoped_view import (
ResourceScope,
ScopedBackendSet,
ScopedBackendView,
ScopeViolationError,
create_scoped_backend_set,
)
from cleveragents.domain.models.acms.strategy import (
BackendSet,
ContextStrategy,
@@ -111,7 +130,6 @@ from cleveragents.domain.models.acms.tiers import (
ActorContextView,
ActorRole,
ContextTier,
ScopedBackendView,
TierBudget,
TieredFragment,
TierMetrics,
@@ -143,6 +161,10 @@ __all__: list[str] = [
"PlanContext",
"PlanDecisionContextStrategy",
"PythonAnalyzer",
"ResourceAliasResolver",
"ResourceScope",
"ScopeViolationError",
"ScopedBackendSet",
"ScopedBackendView",
"SemanticEmbeddingStrategy",
"SimpleKeywordStrategy",
@@ -158,4 +180,8 @@ __all__: list[str] = [
"UKOTriple",
"VectorBackend",
"VectorResult",
"create_scoped_backend_set",
"resolve_resource_scope",
"validate_project_scope",
"validate_resource_scope",
]
@@ -0,0 +1,325 @@
"""Resource scope resolution, alias mapping, and validation guards.
Extracted from ``scoped_view.py`` to keep modules under the 500-line
guideline. This module provides:
| Type / Function | Role |
|----------------------------|-------------------------------------------------|
| ``ResourceAliasResolver`` | Resolves resource aliases to canonical ULIDs |
| ``validate_project_scope`` | Guard for mixed-project scope requests |
| ``validate_resource_scope``| Guard for out-of-scope resource requests |
| ``resolve_resource_scope`` | Builds a ``ResourceScope`` with allow/deny list |
Based on ``docs/specification.md`` > ACMS > Scoped Views and Plan
Subgraph Projection (lines 42501--42566) and Forgejo issue #193.
"""
from __future__ import annotations
from collections.abc import Callable
from typing import TYPE_CHECKING
import structlog
from cleveragents.domain.models.acms.scoped_view import (
ResourceScope,
ScopeViolationError,
)
if TYPE_CHECKING:
from cleveragents.domain.models.core.project import (
NamespacedProject,
)
logger = structlog.get_logger(__name__)
# ---------------------------------------------------------------------------
# ResourceAliasResolver
# ---------------------------------------------------------------------------
class ResourceAliasResolver:
"""Resolves resource aliases to resource IDs with uniqueness validation.
Stateless helper that operates on project data passed to each
method. Used during scope resolution to translate user-facing
resource names and aliases into canonical ULID identifiers.
Based on ``docs/specification.md`` LinkedResource alias field
(line 19715) and Forgejo issue #193.
"""
def resolve(
self,
project: NamespacedProject,
alias_or_id: str,
) -> str | None:
"""Resolve a resource alias or ID to a canonical resource ID.
Checks aliases first (exact match), then falls back to
resource_id matching.
Args:
project: The project containing linked resources.
alias_or_id: An alias or resource ULID to resolve.
Returns:
The canonical resource ULID, or ``None`` if not found.
Raises:
ValueError: If *alias_or_id* is empty or whitespace-only.
"""
if not alias_or_id or not alias_or_id.strip():
raise ValueError("alias_or_id cannot be empty")
# Try alias match first
by_alias = project.get_linked_resource_by_alias(alias_or_id)
if by_alias is not None:
return by_alias.resource_id
# Try direct resource_id match
by_id = project.get_linked_resource(alias_or_id)
if by_id is not None:
return by_id.resource_id
return None
def resolve_many(
self,
project: NamespacedProject,
names: tuple[str, ...],
) -> frozenset[str]:
"""Resolve multiple aliases/IDs to canonical resource IDs.
Unresolvable names are logged and skipped.
Args:
project: The project containing linked resources.
names: Aliases or resource ULIDs to resolve.
Returns:
Frozenset of resolved resource ULIDs.
"""
resolved: set[str] = set()
for name in names:
rid = self.resolve(project, name)
if rid is not None:
resolved.add(rid)
else:
logger.warning(
"alias.unresolved",
alias=name,
project=project.namespaced_name,
)
return frozenset(resolved)
def validate_uniqueness(
self,
project: NamespacedProject,
) -> list[str]:
"""Validate that all aliases in a project are unique.
Args:
project: The project to validate.
Returns:
List of error messages for duplicate aliases (empty if valid).
"""
seen: dict[str, list[str]] = {}
for lr in project.linked_resources:
if lr.alias is not None:
seen.setdefault(lr.alias, []).append(lr.resource_id)
errors: list[str] = []
for alias, resource_ids in seen.items():
if len(resource_ids) > 1:
errors.append(
f"Duplicate alias '{alias}' maps to resources: "
f"{', '.join(sorted(resource_ids))}"
)
return errors
# ---------------------------------------------------------------------------
# Mixed project scope guard
# ---------------------------------------------------------------------------
def validate_project_scope(
requested_projects: frozenset[str],
available_projects: frozenset[str],
) -> None:
"""Validate that all requested projects are available.
Raises ``ScopeViolationError`` if any requested project is not
in the available set. This guards against context requests that
span unlinked projects.
Args:
requested_projects: Project names the request references.
available_projects: Project names available in the scope.
Raises:
ScopeViolationError: If any requested project is not available.
ValueError: If either argument is empty.
"""
if not requested_projects:
raise ValueError("requested_projects must be non-empty")
if not available_projects:
raise ValueError("available_projects must be non-empty")
out_of_scope = requested_projects - available_projects
if out_of_scope:
sorted_bad = sorted(out_of_scope)
logger.warning(
"scope.project_violation",
out_of_scope=sorted_bad,
available=sorted(available_projects),
)
raise ScopeViolationError(
f"Projects not in scope: {', '.join(sorted_bad)}. "
f"Available: {', '.join(sorted(available_projects))}",
scope_project_names=tuple(sorted(available_projects)),
)
def validate_resource_scope(
requested_resource_ids: frozenset[str],
scope: ResourceScope,
) -> None:
"""Validate that all requested resource IDs are within scope.
Raises ``ScopeViolationError`` if any resource ID is not in the
scope's ``resource_ids``.
Args:
requested_resource_ids: Resource ULIDs the request references.
scope: The active resource scope.
Raises:
ScopeViolationError: If any resource is out of scope.
"""
if not requested_resource_ids:
return
out_of_scope = requested_resource_ids - scope.resource_ids
if out_of_scope:
sorted_bad = sorted(out_of_scope)
logger.warning(
"scope.resource_violation",
out_of_scope=sorted_bad,
scope_resource_count=len(scope.resource_ids),
)
raise ScopeViolationError(
f"Resources not in scope: {', '.join(sorted_bad)}",
resource_ids=tuple(sorted_bad),
scope_project_names=tuple(sorted(scope.project_names)),
)
# ---------------------------------------------------------------------------
# Allowlist/denylist resolution
# ---------------------------------------------------------------------------
def resolve_resource_scope(
projects: list[NamespacedProject],
*,
include_resources: tuple[str, ...] = (),
exclude_resources: tuple[str, ...] = (),
include_paths: tuple[str, ...] = (),
exclude_paths: tuple[str, ...] = (),
temporal_scope: str = "current",
registry_lookup: Callable[[str], frozenset[str]] | None = None,
) -> ResourceScope:
"""Resolve a ``ResourceScope`` from projects and filter arguments.
Collects all resource IDs from linked resources across the given
projects, then applies allowlist/denylist filtering using alias
resolution.
When *registry_lookup* is provided, each resource ID is expanded
to include its DAG descendants (per ``docs/specification.md``
lines 42517--42521). The callback receives a resource ULID and
returns the frozenset of descendant ULIDs. When ``None``, no
DAG expansion is performed (flat ULID-set mode).
Args:
projects: Projects whose linked resources form the base scope.
include_resources: Resource allowlist (names/aliases).
Empty means include all.
exclude_resources: Resource denylist (names/aliases).
Applied after allowlist.
include_paths: Path glob allowlist.
exclude_paths: Path glob denylist.
temporal_scope: One of ``"current"``, ``"recent"``, ``"all"``.
registry_lookup: Optional callback to expand a resource ULID
to its DAG descendants. Signature:
``(resource_id: str) -> frozenset[str]``.
Returns:
A ``ResourceScope`` with the resolved resource IDs and filters.
Raises:
ValueError: If *projects* is empty.
"""
if not projects:
raise ValueError("projects must be non-empty")
resolver = ResourceAliasResolver()
all_resource_ids: set[str] = set()
project_names: set[str] = set()
for project in projects:
project_names.add(project.namespaced_name)
for lr in project.linked_resources:
all_resource_ids.add(lr.resource_id)
# DAG expansion: include child resources per spec §42517-42521
if registry_lookup is not None:
expanded: set[str] = set()
for rid in all_resource_ids:
expanded.add(rid)
expanded.update(registry_lookup(rid))
all_resource_ids = expanded
# Apply allowlist: resolve names/aliases to resource IDs
if include_resources:
allowed: set[str] = set()
for project in projects:
allowed.update(resolver.resolve_many(project, include_resources))
all_resource_ids &= allowed
# Apply denylist: resolve names/aliases to resource IDs to exclude
if exclude_resources:
denied: set[str] = set()
for project in projects:
denied.update(resolver.resolve_many(project, exclude_resources))
all_resource_ids -= denied
if not all_resource_ids:
logger.warning(
"scope.empty_after_filtering",
project_names=sorted(project_names),
include_resources=include_resources,
exclude_resources=exclude_resources,
)
return ResourceScope(
resource_ids=frozenset(all_resource_ids),
project_names=frozenset(project_names),
temporal_scope=temporal_scope,
include_resources=include_resources,
exclude_resources=exclude_resources,
include_paths=include_paths,
exclude_paths=exclude_paths,
)
__all__: list[str] = [
"ResourceAliasResolver",
"resolve_resource_scope",
"validate_project_scope",
"validate_resource_scope",
]
@@ -0,0 +1,473 @@
"""Scoped backend view filtering for project-resource isolation.
Core types: ``ResourceScope``, ``ScopeViolationError``,
``ScopedBackendView``, ``ScopedBackendSet``, ``create_scoped_backend_set``.
Resolution helpers live in ``scope_resolution.py``.
Based on ``docs/specification.md`` lines 42501--42566 and issue #193.
"""
from __future__ import annotations
from pathlib import PurePath
import structlog
from pydantic import BaseModel, ConfigDict, Field, PrivateAttr, field_validator
from cleveragents.domain.models.acms.backends import (
GraphBackend,
GraphResult,
TextBackend,
TextResult,
VectorBackend,
VectorResult,
)
logger = structlog.get_logger(__name__)
# ---------------------------------------------------------------------------
# Exceptions
# ---------------------------------------------------------------------------
class ScopeViolationError(Exception):
"""Raised when a context request references resources outside scope.
Attributes:
resource_ids: The out-of-scope resource identifiers.
scope_project_names: The project names that define the scope.
"""
def __init__(
self,
message: str,
*,
resource_ids: tuple[str, ...] = (),
scope_project_names: tuple[str, ...] = (),
) -> None:
super().__init__(message)
self.resource_ids = resource_ids
self.scope_project_names = scope_project_names
# ---------------------------------------------------------------------------
# ResourceScope
# ---------------------------------------------------------------------------
class ResourceScope(BaseModel):
"""Resolved set of resources and projects visible to a plan.
Immutable once created. Based on ``specification.md`` lines 42510--42525.
"""
resource_ids: frozenset[str] = Field(
...,
description="Resource ULIDs in scope (empty = no resources after filtering)",
)
project_names: frozenset[str] = Field(
...,
min_length=1,
description="Namespaced project names in scope",
)
temporal_scope: str = Field(
default="current",
description="Temporal scope for UKO node resolution",
pattern=r"^(current|recent|all)$",
)
include_resources: tuple[str, ...] = Field(
default=(),
max_length=256,
description="Resource allowlist (names or aliases). Empty = all.",
)
exclude_resources: tuple[str, ...] = Field(
default=(),
max_length=256,
description="Resource denylist (names or aliases). Applied after allowlist.",
)
include_paths: tuple[str, ...] = Field(
default=(),
max_length=256,
description="Path glob allowlist. Empty = all.",
)
exclude_paths: tuple[str, ...] = Field(
default=(),
max_length=256,
description="Path glob denylist. Applied after allowlist.",
)
model_config = ConfigDict(frozen=True)
def contains_resource(self, resource_id: str) -> bool:
"""Return True if *resource_id* is within this scope.
Args:
resource_id: A resource ULID to check.
Returns:
Whether the resource is in scope.
"""
return resource_id in self.resource_ids
def contains_project(self, project_name: str) -> bool:
"""Return True if *project_name* is within this scope.
Args:
project_name: A namespaced project name to check.
Returns:
Whether the project is in scope.
"""
return project_name in self.project_names
def matches_path(self, path: str) -> bool:
"""Return True if *path* passes include/exclude path filters.
Uses ``PurePath.full_match()`` (requires Python >=3.12; this
project requires >=3.13 per ``pyproject.toml``).
"""
pp = PurePath(path)
if self.include_paths and not any(
pp.full_match(pat) for pat in self.include_paths
):
return False
return not (
self.exclude_paths and any(pp.full_match(pat) for pat in self.exclude_paths)
)
# ---------------------------------------------------------------------------
# ScopedBackendView
# ---------------------------------------------------------------------------
class ScopedBackendView(BaseModel):
"""Wraps a backend to restrict queries to a resource scope.
Operates at two levels: backend-level (injects ``scope`` into
queries) and fragment-level (``is_visible()`` filters by project
and resource). Based on ``specification.md`` lines 42548--42566.
"""
allowed_projects: frozenset[str] = Field(
...,
min_length=1,
description="Project names the actor is permitted to access",
)
resource_ids: frozenset[str] = Field(
default=frozenset(),
description="Resource ULIDs in scope (empty = allow all for backward compat)",
)
denied_resource_ids: frozenset[str] = Field(
default=frozenset(),
description="Resource ULIDs explicitly denied (denylist)",
)
_effective_scope_cached: frozenset[str] = PrivateAttr(
default=frozenset(),
)
model_config = ConfigDict(
frozen=True,
)
def model_post_init(self, __context: object) -> None:
"""Pre-compute effective scope (model is frozen after init)."""
if not self.resource_ids:
self._effective_scope_cached = frozenset()
else:
self._effective_scope_cached = self.resource_ids - self.denied_resource_ids
@field_validator("allowed_projects", mode="before")
@classmethod
def _coerce_allowed_projects(
cls: type[ScopedBackendView],
v: frozenset[str] | list[str] | set[str] | tuple[str, ...],
) -> frozenset[str]:
"""Accept list/set/tuple and convert to frozenset."""
if isinstance(v, (list, set, tuple)):
return frozenset(v)
return v
@field_validator("resource_ids", mode="before")
@classmethod
def _coerce_resource_ids(
cls: type[ScopedBackendView],
v: frozenset[str] | list[str] | set[str] | tuple[str, ...],
) -> frozenset[str]:
"""Accept list/set/tuple and convert to frozenset."""
if isinstance(v, (list, set, tuple)):
return frozenset(v)
return v
@field_validator("denied_resource_ids", mode="before")
@classmethod
def _coerce_denied_resource_ids(
cls: type[ScopedBackendView],
v: frozenset[str] | list[str] | set[str] | tuple[str, ...],
) -> frozenset[str]:
"""Accept list/set/tuple and convert to frozenset."""
if isinstance(v, (list, set, tuple)):
return frozenset(v)
return v
# -- Resource scope checks -----------------------------------------------
def is_resource_in_scope(self, resource_id: str) -> bool:
"""Return True if *resource_id* is allowed by this view.
A resource is in scope when:
1. It is in ``resource_ids`` (or ``resource_ids`` is empty), AND
2. It is NOT in ``denied_resource_ids``.
Args:
resource_id: A resource ULID to check.
Returns:
Whether the resource passes the scope filter.
"""
if resource_id in self.denied_resource_ids:
return False
if self.resource_ids:
return resource_id in self.resource_ids
return True
def effective_scope(self) -> frozenset[str]:
"""Return ``resource_ids - denied_resource_ids`` (cached at init).
Empty frozenset when ``resource_ids`` is empty (project-only mode).
"""
return self._effective_scope_cached
# -- Fragment-level filtering (backward compat with tier service) ---------
def is_visible(self, fragment: object) -> bool:
"""Return True if *fragment* belongs to an allowed project.
Fragments with an empty ``project_name`` are **excluded** by
default to prevent information leakage.
When ``resource_ids`` is populated (ULID-level filtering is
active), fragments **must** have a non-empty ``resource_id``
that passes ``is_resource_in_scope()``. Fragments with an
empty ``resource_id`` are denied to enforce resource-level
isolation (deny-by-default).
Args:
fragment: A ``TieredFragment`` or any object with
``project_name`` and optionally ``resource_id`` attrs.
Returns:
Whether the fragment is visible in this view.
"""
project_name: str = getattr(fragment, "project_name", "")
if not project_name:
return False
if project_name not in self.allowed_projects:
return False
# Resource-level filtering when resource_ids are set
resource_id: str = getattr(fragment, "resource_id", "")
if self.resource_ids:
# Deny-by-default: fragments without a resource_id are
# excluded when resource-level filtering is active.
if not resource_id:
return False
return self.is_resource_in_scope(resource_id)
return True
# -- Backend-level query proxying ----------------------------------------
def _require_effective_scope(self) -> frozenset[str]:
"""Return the effective scope or raise if all resources are denied.
Returns an empty frozenset in project-only mode (no resource
filter). Raises ``ScopeViolationError`` when ``resource_ids``
is populated but all entries are denied.
"""
scope = self.effective_scope()
if self.resource_ids and not scope:
logger.warning(
"scope.empty_after_denylist",
allowed_projects=sorted(self.allowed_projects),
)
raise ScopeViolationError(
"All resources in scope are denied",
scope_project_names=tuple(sorted(self.allowed_projects)),
)
return scope
def search_text(
self,
backend: TextBackend,
query: str,
*,
max_results: int = 20,
) -> list[TextResult]:
"""Search a text backend with scope injected."""
scope = self._require_effective_scope()
return backend.search(query, scope=scope, max_results=max_results)
def search_vector(
self,
backend: VectorBackend,
embedding: list[float],
*,
top_k: int = 20,
) -> list[VectorResult]:
"""Search a vector backend with scope injected."""
scope = self._require_effective_scope()
return backend.similarity_search(embedding, scope=scope, top_k=top_k)
def search_graph(
self,
backend: GraphBackend,
query: str,
) -> GraphResult:
"""Execute a SPARQL query on a graph backend with scope injected."""
scope = self._require_effective_scope()
return backend.sparql_query(query, scope=scope)
@classmethod
def from_resource_scope(cls, scope: ResourceScope) -> ScopedBackendView:
"""Create a ``ScopedBackendView`` from a ``ResourceScope``.
.. note::
``denied_resource_ids`` is **not** forwarded because
``resolve_resource_scope()`` already subtracts denied IDs
from ``resource_ids`` before constructing the
``ResourceScope``. The scope's ``resource_ids`` is
therefore already the effective (post-deny) set.
Args:
scope: The resolved resource scope.
Returns:
A new ``ScopedBackendView`` configured with the scope's
resource IDs and project names.
"""
return cls(
allowed_projects=scope.project_names,
resource_ids=scope.resource_ids,
)
# ---------------------------------------------------------------------------
# ScopedBackendSet
# ---------------------------------------------------------------------------
class ScopedBackendSet(BaseModel):
"""Bundle of scoped backend views for text, vector, and graph.
Created by ``create_scoped_backend_set()`` during context assembly.
Based on ``specification.md`` line 43050.
"""
view: ScopedBackendView = Field(
...,
description="The scoped view enforcing resource isolation",
)
text_backend: TextBackend | None = Field(
default=None,
description="Text search backend (None if unavailable)",
)
vector_backend: VectorBackend | None = Field(
default=None,
description="Vector search backend (None if unavailable)",
)
graph_backend: GraphBackend | None = Field(
default=None,
description="Graph query backend (None if unavailable)",
)
model_config = ConfigDict(arbitrary_types_allowed=True)
def search_text(
self,
query: str,
*,
max_results: int = 20,
) -> list[TextResult]:
"""Search text backend with scope enforcement.
Args:
query: Natural-language or keyword query.
max_results: Maximum results to return.
Returns:
Scoped text results (empty list if no text backend).
"""
if self.text_backend is None:
return []
return self.view.search_text(self.text_backend, query, max_results=max_results)
def search_vector(
self,
embedding: list[float],
*,
top_k: int = 20,
) -> list[VectorResult]:
"""Search vector backend with scope enforcement.
Args:
embedding: Query embedding vector.
top_k: Maximum results to return.
Returns:
Scoped vector results.
"""
if self.vector_backend is None:
return []
return self.view.search_vector(self.vector_backend, embedding, top_k=top_k)
def search_graph(
self,
query: str,
) -> GraphResult:
"""Execute SPARQL query with scope enforcement.
Args:
query: SPARQL query string.
Returns:
Scoped graph results.
"""
if self.graph_backend is None:
return GraphResult()
return self.view.search_graph(self.graph_backend, query)
def create_scoped_backend_set(
scope: ResourceScope,
*,
text_backend: TextBackend | None = None,
vector_backend: VectorBackend | None = None,
graph_backend: GraphBackend | None = None,
) -> ScopedBackendSet:
"""Create a ``ScopedBackendSet`` from a ``ResourceScope``."""
view = ScopedBackendView.from_resource_scope(scope)
logger.info(
"scoped_backend_set.created",
project_names=sorted(scope.project_names),
resource_count=len(scope.resource_ids),
)
return ScopedBackendSet(
view=view,
text_backend=text_backend,
vector_backend=vector_backend,
graph_backend=graph_backend,
)
__all__: list[str] = [
"ResourceScope",
"ScopeViolationError",
"ScopedBackendSet",
"ScopedBackendView",
"create_scoped_backend_set",
]
+17 -30
View File
@@ -255,35 +255,22 @@ class TierMetrics(BaseModel):
# ---------------------------------------------------------------------------
# ScopedBackendView
# ScopedBackendView — re-exported from scoped_view module
# ---------------------------------------------------------------------------
# The full implementation (backend wrapping, resource-level filtering,
# allowlist/denylist, scope validation) lives in ``scoped_view.py``.
# This re-export preserves backward compatibility with existing imports
# from this module.
from cleveragents.domain.models.acms.scoped_view import ( # noqa: E402
ScopedBackendView,
)
class ScopedBackendView(BaseModel):
"""Project-scoped filter ensuring actors never see resources outside scope.
Enforces project isolation by restricting fragment visibility to
only those fragments whose ``project_name`` matches one of the
allowed project names.
"""
allowed_projects: list[str] = Field(
...,
min_length=1,
description="Project names the actor is permitted to access",
)
def is_visible(self, fragment: TieredFragment) -> bool:
"""Return True if *fragment* belongs to an allowed project.
Fragments with an empty ``project_name`` are **excluded** by
default to prevent information leakage.
"""
if not fragment.project_name:
return False
return fragment.project_name in self.allowed_projects
model_config = ConfigDict(
str_strip_whitespace=True,
validate_assignment=True,
)
__all__: list[str] = [
"ActorContextView",
"ActorRole",
"ContextTier",
"ScopedBackendView",
"TierBudget",
"TierMetrics",
"TieredFragment",
]