cleveragents/cleveragents-core
4
0
Fork 3
Code Issues 6.0k Pull Requests 115 Actions Packages Projects Releases Wiki Activity

feat(domain): add spec-aligned Resource and Project models #51

Merged
brent.edwards merged 2 commits from feat/B1-v2-project-data-models into master 2026-02-13 22:53:40 +00:00
Conversation 1 Commits 2 Files Changed 10
+2414 -68
10 changed files with 2414 additions and 68 deletions
Download Patch File Download Diff File Expand all files Collapse all files
benchmarks/plan_generation_benchmark.py
+1 -1
View File
@@ -18,7 +18,7 @@ from cleveragents.domain.models.core import (
PlanStatus,
Project,
)
from cleveragents.domain.models.core.project import ProjectSettings
from cleveragents.domain.models.core.project_legacy import ProjectSettings
def _sample_project() -> Project:
features/namespaced_project_model.feature
+258
View File
@@ -0,0 +1,258 @@
Feature: Namespaced Project Domain Model
As a developer
I want spec-aligned project domain models
So that projects use namespaced names with linked resources
# parse_namespaced_name
Scenario: Parse bare name defaults to local namespace
When I parse the project namespaced name "my-project"
Then the nsproject parsed namespace should be "local"
And the nsproject parsed name should be "my-project"
And the nsproject parsed server should be empty
Scenario: Parse name with explicit namespace
When I parse the project namespaced name "freemo/code-coverage"
Then the nsproject parsed namespace should be "freemo"
And the nsproject parsed name should be "code-coverage"
And the nsproject parsed server should be empty
Scenario: Parse name with server qualifier
When I parse the project namespaced name "dev:freemo/code-coverage"
Then the nsproject parsed server should be "dev"
And the nsproject parsed namespace should be "freemo"
And the nsproject parsed name should be "code-coverage"
Scenario: Parse local namespace
When I parse the project namespaced name "local/my-project"
Then the nsproject parsed namespace should be "local"
And the nsproject parsed name should be "my-project"
Scenario: ParsedName qualified_name with server
When I parse the project namespaced name "prod:myorg/api"
Then the nsproject qualified name should be "prod:myorg/api"
Scenario: ParsedName qualified_name without server
When I parse the project namespaced name "myorg/api"
Then the nsproject qualified name should be "myorg/api"
Scenario: ParsedName namespaced_name
When I parse the project namespaced name "dev:myorg/api"
Then the nsproject namespaced name should be "myorg/api"
Scenario: ParsedName is_local for bare name
When I parse the project namespaced name "my-project"
Then the nsproject parsed name should be local
Scenario: ParsedName is_remote with server
When I parse the project namespaced name "dev:myorg/api"
Then the nsproject parsed name should be remote
Scenario: ParsedName is_remote with non-local namespace
When I parse the project namespaced name "freemo/api"
Then the nsproject parsed name should be remote
Scenario: Reject empty name
When I try to parse an empty project namespaced name
Then a nsproject validation error should be raised
Scenario: Reject reserved namespace system
When I try to parse the project namespaced name "system/my-project"
Then a nsproject validation error should be raised
Scenario: Reject reserved namespace admin
When I try to parse the project namespaced name "admin/my-project"
Then a nsproject validation error should be raised
Scenario: Reject provider namespace openai
When I try to parse the project namespaced name "openai/my-project"
Then a nsproject validation error should be raised
Scenario: Reject provider namespace anthropic
When I try to parse the project namespaced name "anthropic/my-model"
Then a nsproject validation error should be raised
Scenario: Reject invalid characters in name
When I try to parse the project namespaced name "my@project"
Then a nsproject validation error should be raised
# TemporalScope enum
Scenario: TemporalScope has three values
Then the nsproject TemporalScope enum should have values "current, recent, all"
# ContextConfig model
Scenario: ContextConfig defaults
Given a default ContextConfig
Then the nsproject ctx max_file_size should be 1000000
And the nsproject ctx max_total_size should be 52428800
And the nsproject ctx indexing_strategy should be "full_text"
And the nsproject ctx chunking_policy should be "smart"
And the nsproject ctx chunk_size should be 1000
And the nsproject ctx summarize should be true
And the nsproject ctx auto_refresh should be true
And the nsproject ctx temporal_scope should be "current"
And the nsproject ctx retention_policy should be empty
Scenario: ContextConfig merges default ignore patterns
Given a ContextConfig with custom ignore patterns
Then the nsproject ctx ignore patterns should include defaults
And the nsproject ctx ignore patterns should include the custom pattern
Scenario: ContextConfig with memory tier settings
Given a ContextConfig with hot_max_tokens 4096 and warm_max_decisions 50
Then the nsproject ctx hot_max_tokens should be 4096
And the nsproject ctx warm_max_decisions should be 50
Scenario: ContextConfig is frozen
Given a default ContextConfig
When I try to mutate a ContextConfig field
Then a nsproject validation error should be raised
Scenario: ContextConfig rejects zero max_file_size
When I try to create a ContextConfig with max_file_size 0
Then a nsproject validation error should be raised
Scenario: ContextConfig rejects negative chunk_size
When I try to create a ContextConfig with chunk_size -1
Then a nsproject validation error should be raised
# LinkedResource model
Scenario: Create a LinkedResource with defaults
Given a LinkedResource with resource_id "01HZQX7K8M3N4P5R6S7T8V9W0X"
Then the nsproject linked rid should be "01HZQX7K8M3N4P5R6S7T8V9W0X"
And the nsproject linked read_only should be false
And the nsproject linked alias should be empty
And the nsproject linked_at should be set
Scenario: Create a LinkedResource with alias
Given a LinkedResource with id "01HZQX7K8M3N4P5R6S7T8V9W0X" and alias "main-repo"
Then the nsproject linked alias should be "main-repo"
Scenario: LinkedResource rejects invalid ULID
When I try to create a LinkedResource with resource_id "invalid"
Then a nsproject validation error should be raised
Scenario: LinkedResource rejects empty alias
When I try to create a LinkedResource with empty alias
Then a nsproject validation error should be raised
Scenario: LinkedResource is frozen
Given a LinkedResource with resource_id "01HZQX7K8M3N4P5R6S7T8V9W0X"
When I try to mutate a LinkedResource field
Then a nsproject validation error should be raised
# NamespacedProject model
Scenario: Create a minimal NamespacedProject
Given a NamespacedProject with name "my-project"
Then the nsproject name should be "my-project"
And the nsproject namespace should be "local"
And the nsproject server should be empty
And the nsproject namespaced name should be "local/my-project"
And the nsproject qualified name should be "local/my-project"
And the nsproject linked resources should be empty
And the nsproject created_at should be set
And the nsproject updated_at should be set
Scenario: Create a NamespacedProject with namespace
Given a project "api" under namespace "freemo"
Then the nsproject namespace should be "freemo"
And the nsproject namespaced name should be "freemo/api"
Scenario: Create a NamespacedProject with server
Given a project "api" under namespace "freemo" on server "dev"
Then the nsproject server should be "dev"
And the nsproject qualified name should be "dev:freemo/api"
Scenario: NamespacedProject is_local
Given a NamespacedProject with name "my-project"
Then the nsproject should be local
Scenario: NamespacedProject is_remote with namespace
Given a project "api" under namespace "freemo"
Then the nsproject should be remote
Scenario: NamespacedProject is_remote with server
Given a project "api" under namespace "freemo" on server "dev"
Then the nsproject should be remote
# NamespacedProject - validation
Scenario: NamespacedProject rejects invalid name
When I try to create a NamespacedProject with name "123invalid"
Then a nsproject validation error should be raised
Scenario: NamespacedProject rejects reserved namespace
When I try to create a NamespacedProject with namespace "system"
Then a nsproject validation error should be raised
Scenario: NamespacedProject rejects provider namespace
When I try to create a NamespacedProject with namespace "openai"
Then a nsproject validation error should be raised
Scenario: NamespacedProject rejects duplicate linked resources
When I try to create a NamespacedProject with duplicate linked resources
Then a nsproject validation error should be raised
# NamespacedProject - domain methods
Scenario: Link a resource to a project
Given a NamespacedProject with name "my-project"
When I link resource "01HZQX7K8M3N4P5R6S7T8V9W0X" to the project
Then the nsproject should have 1 linked resource
And the nsproject first linked resource_id should be "01HZQX7K8M3N4P5R6S7T8V9W0X"
Scenario: Link a resource with read_only and alias
Given a NamespacedProject with name "my-project"
When I link resource "01HZQX7K8M3N4P5R6S7T8V9W0X" with read_only true and alias "docs"
Then the nsproject should have 1 linked resource
And the nsproject first linked resource should be read_only
And the nsproject first linked resource alias should be "docs"
Scenario: Unlink a resource from a project
Given a NamespacedProject with name "my-project"
And I link resource "01HZQX7K8M3N4P5R6S7T8V9W0X" to the project
When I unlink resource "01HZQX7K8M3N4P5R6S7T8V9W0X" from the project
Then the nsproject linked resources should be empty
Scenario: Link duplicate resource fails
Given a NamespacedProject with name "my-project"
And I link resource "01HZQX7K8M3N4P5R6S7T8V9W0X" to the project
When I try to link resource "01HZQX7K8M3N4P5R6S7T8V9W0X" again
Then a nsproject validation error should be raised
Scenario: Unlink non-existent resource fails
Given a NamespacedProject with name "my-project"
When I try to unlink resource "01HZQX7K8M3N4P5R6S7T8V9W0X" from the project
Then a nsproject validation error should be raised
Scenario: Get linked resource by resource_id
Given a NamespacedProject with name "my-project"
And I link resource "01HZQX7K8M3N4P5R6S7T8V9W0X" to the project
When I get linked resource "01HZQX7K8M3N4P5R6S7T8V9W0X" from the project
Then the nsproject retrieved link should not be empty
Scenario: Get linked resource by alias
Given a NamespacedProject with name "my-project"
And I link resource "01HZQX7K8M3N4P5R6S7T8V9W0X" with read_only false and alias "main"
When I get linked resource by alias "main" from the project
Then the nsproject retrieved link should not be empty
And the nsproject retrieved link alias should be "main"
Scenario: Get non-existent linked resource returns None
Given a NamespacedProject with name "my-project"
When I get linked resource "01HZQX7K8M3N4P5R6S7T8V9W0X" from the project
Then the nsproject retrieved link should be empty
Scenario: Link resource with empty resource_id fails
Given a NamespacedProject with name "my-project"
When I try to link an empty resource_id to the project
Then a nsproject validation error should be raised
Scenario: Unlink resource with empty resource_id fails
Given a NamespacedProject with name "my-project"
When I try to unlink an empty resource_id from the project
Then a nsproject validation error should be raised
features/resource_registry_model.feature
+171
View File
@@ -0,0 +1,171 @@
Feature: Resource Registry Domain Model
As a developer
I want spec-aligned Resource Registry domain models
So that resources are tracked with proper types, capabilities, and relationships
# PhysVirt enum
Scenario: PhysVirt has exactly two values
Then the PhysVirt enum should have values "physical, virtual"
Scenario: PhysVirt physical value
Given a PhysVirt value of "physical"
Then the PhysVirt string value should be "physical"
Scenario: PhysVirt virtual value
Given a PhysVirt value of "virtual"
Then the PhysVirt string value should be "virtual"
Scenario: PhysVirt rejects invalid value
When I try to create a PhysVirt with value "hybrid"
Then a resource validation error should be raised
# SandboxStrategy enum
Scenario: SandboxStrategy has exactly five values
Then the SandboxStrategy enum should have values "git_worktree, copy_on_write, transaction_rollback, snapshot, none"
Scenario Outline: SandboxStrategy accepts valid values
Given a SandboxStrategy value of "<strategy>"
Then the SandboxStrategy string value should be "<strategy>"
Examples:
| strategy |
| git_worktree |
| copy_on_write |
| transaction_rollback |
| snapshot |
| none |
Scenario: SandboxStrategy rejects overlay
When I try to create a SandboxStrategy with value "overlay"
Then a resource validation error should be raised
Scenario: SandboxStrategy rejects versioning
When I try to create a SandboxStrategy with value "versioning"
Then a resource validation error should be raised
# ResourceCapabilities model
Scenario: ResourceCapabilities defaults
Given default ResourceCapabilities
Then the capability readable should be true
And the capability writable should be true
And the capability sandboxable should be true
And the capability checkpointable should be false
Scenario: ResourceCapabilities with custom values
Given ResourceCapabilities with readable false and checkpointable true
Then the capability readable should be false
And the capability writable should be true
And the capability sandboxable should be true
And the capability checkpointable should be true
Scenario: ResourceCapabilities is frozen
Given default ResourceCapabilities
When I try to mutate a ResourceCapabilities field
Then a resource validation error should be raised
# Resource model - creation
Scenario: Create a minimal physical resource
Given a Resource with resource_id "01HZQX7K8M3N4P5R6S7T8V9W0X" type "git-checkout" and classification "physical"
Then the resource resource_id should be "01HZQX7K8M3N4P5R6S7T8V9W0X"
And the resource type name should be "git-checkout"
And the resource classification should be "physical"
And the resource name should be empty
And the resource description should be empty
And the resource properties should be empty
And the resource parents should be empty
And the resource children should be empty
And the resource linked_projects should be empty
And the resource created_at should be set
And the resource updated_at should be set
Scenario: Create a full resource with all fields
Given a full Resource with id "01HZQX7K8M3N4P5R6S7T8V9W0X" name "local/my-repo" type "git-checkout" classification "physical" description "Main repo"
Then the resource name should be "local/my-repo"
And the resource description should be "Main repo"
Scenario: Resource with sandbox strategy override
Given a Resource with sandbox strategy "git_worktree"
Then the resource sandbox strategy should be "git_worktree"
Scenario: Resource with capabilities
Given a Resource with sandboxable false and checkpointable true
Then the resource can_sandbox should be false
And the resource can_checkpoint should be true
# Resource model - validation
Scenario: Resource rejects invalid ULID
When I try to create a Resource with resource_id "invalid-ulid"
Then a resource validation error should be raised
Scenario: Resource rejects empty resource_type_name
When I try to create a Resource with empty resource_type_name
Then a resource validation error should be raised
Scenario: Resource rejects empty string name
When I try to create a Resource with empty string name
Then a resource validation error should be raised
Scenario: Resource accepts None name
Given a Resource with resource_id "01HZQX7K8M3N4P5R6S7T8V9W0X" type "git-checkout" and classification "physical"
Then the resource name should be empty
# Resource model - domain properties
Scenario: Physical resource is_physical
Given a Resource with resource_id "01HZQX7K8M3N4P5R6S7T8V9W0X" type "git-checkout" and classification "physical"
Then the resource is_physical should be true
And the resource is_virtual should be false
Scenario: Virtual resource is_virtual
Given a Resource with resource_id "01HZQX7K8M3N4P5R6S7T8V9W0X" type "git" and classification "virtual"
Then the resource is_physical should be false
And the resource is_virtual should be true
Scenario: Read-only resource
Given a Resource with writable false
Then the resource is_read_only should be true
Scenario: Writable resource
Given a Resource with resource_id "01HZQX7K8M3N4P5R6S7T8V9W0X" type "git-checkout" and classification "physical"
Then the resource is_read_only should be false
# Resource model - frozen
Scenario: Resource is frozen
Given a Resource with resource_id "01HZQX7K8M3N4P5R6S7T8V9W0X" type "git-checkout" and classification "physical"
When I try to mutate a Resource field
Then a resource validation error should be raised
# Resource model - with_updated_at
Scenario: with_updated_at returns new instance
Given a Resource with resource_id "01HZQX7K8M3N4P5R6S7T8V9W0X" type "git-checkout" and classification "physical"
When I call with_updated_at on the resource
Then the new resource should have a different updated_at
And the new resource should have the same resource_id
# Resource model - content_hash and location
Scenario: Resource with content_hash
Given a Resource with content_hash "sha256:abc123"
Then the resource content_hash should be "sha256:abc123"
Scenario: Resource with location
Given a Resource with location "/home/user/repos/main"
Then the resource location should be "/home/user/repos/main"
Scenario: Resource rejects empty location string
When I try to create a Resource with empty location
Then a resource validation error should be raised
# Resource model - parents and children
Scenario: Resource with parents and children
Given a Resource with parents and children
Then the resource should have 1 parent
And the resource should have 2 children
features/steps/auto_debug_cli_coverage_steps.py
+1 -1
View File
@@ -18,7 +18,7 @@ from cleveragents.cli.commands.auto_debug import (
app as auto_debug_app,
)
from cleveragents.core.exceptions import CleverAgentsError, PlanError
from cleveragents.domain.models.core.project import Project
from cleveragents.domain.models.core.project_legacy import Project
runner = CliRunner()
features/steps/namespaced_project_model_steps.py
+613
View File
@@ -0,0 +1,613 @@
"""Step definitions for Namespaced Project domain model tests.
All assertion steps use 'nsproject' prefix to avoid collisions
with the 128 other step files loaded by Behave globally.
"""
from __future__ import annotations
from behave import given, then, use_step_matcher, when # type: ignore[attr-defined]
from behave.runner import Context
from pydantic import ValidationError
from cleveragents.domain.models.core.project import (
ContextConfig,
LinkedResource,
NamespacedProject,
TemporalScope,
parse_namespaced_name,
)
VALID_ULID = "01HZQX7K8M3N4P5R6S7T8V9W0X"
# -- parse_namespaced_name ---------------------------------------------------
@when('I parse the project namespaced name "{value}"')
def step_parse_project_name(context: Context, value: str) -> None:
"""Parse a namespaced name string."""
context.parsed_name = parse_namespaced_name(value)
@then('the nsproject parsed namespace should be "{expected}"')
def step_check_parsed_namespace(context: Context, expected: str) -> None:
"""Check parsed namespace."""
assert context.parsed_name.namespace == expected, (
f"Expected '{expected}', got '{context.parsed_name.namespace}'"
)
@then('the nsproject parsed name should be "{expected}"')
def step_check_parsed_name(context: Context, expected: str) -> None:
"""Check parsed name."""
assert context.parsed_name.name == expected, (
f"Expected '{expected}', got '{context.parsed_name.name}'"
)
@then("the nsproject parsed server should be empty")
def step_check_parsed_server_empty(context: Context) -> None:
"""Check parsed server is None."""
assert context.parsed_name.server is None, (
f"Expected None, got '{context.parsed_name.server}'"
)
@then('the nsproject parsed server should be "{expected}"')
def step_check_parsed_server(context: Context, expected: str) -> None:
"""Check parsed server."""
assert context.parsed_name.server == expected, (
f"Expected '{expected}', got '{context.parsed_name.server}'"
)
@then('the nsproject qualified name should be "{expected}"')
def step_check_qualified_name(context: Context, expected: str) -> None:
"""Check qualified name from parsed or project context."""
obj = getattr(context, "parsed_name", None) or context.project
assert obj.qualified_name == expected, (
f"Expected '{expected}', got '{obj.qualified_name}'"
)
@then('the nsproject namespaced name should be "{expected}"')
def step_check_namespaced_name(context: Context, expected: str) -> None:
"""Check namespaced_name from parsed or project context."""
obj = getattr(context, "parsed_name", None) or context.project
assert obj.namespaced_name == expected, (
f"Expected '{expected}', got '{obj.namespaced_name}'"
)
@then("the nsproject parsed name should be local")
def step_check_parsed_is_local(context: Context) -> None:
"""Check parsed name is local."""
assert context.parsed_name.is_local, "Expected is_local=True"
@then("the nsproject parsed name should be remote")
def step_check_parsed_is_remote(context: Context) -> None:
"""Check parsed name is remote."""
assert context.parsed_name.is_remote, "Expected is_remote=True"
@when('I try to parse the project namespaced name "{value}"')
def step_try_parse_project_name(context: Context, value: str) -> None:
"""Try to parse a namespaced name, expecting failure."""
try:
parse_namespaced_name(value)
context.project_error = None
except ValueError as e:
context.project_error = e
@when("I try to parse an empty project namespaced name")
def step_try_parse_empty_name(context: Context) -> None:
"""Try to parse an empty string."""
try:
parse_namespaced_name("")
context.project_error = None
except ValueError as e:
context.project_error = e
@then("a nsproject validation error should be raised")
def step_check_project_error(context: Context) -> None:
"""Check that a validation error was raised."""
assert context.project_error is not None, (
"Expected a validation error but none was raised"
)
# -- TemporalScope enum -----------------------------------------------------
@then('the nsproject TemporalScope enum should have values "{values}"')
def step_temporal_scope_values(context: Context, values: str) -> None:
"""Check TemporalScope enum values."""
expected = {v.strip() for v in values.split(",")}
actual = {member.value for member in TemporalScope}
assert actual == expected, f"Expected {expected}, got {actual}"
# -- ContextConfig model -----------------------------------------------------
@given("a default ContextConfig")
def step_default_context_config(context: Context) -> None:
"""Create a default ContextConfig."""
context.context_config = ContextConfig()
@given("a ContextConfig with custom ignore patterns")
def step_context_config_custom_ignores(context: Context) -> None:
"""Create a ContextConfig with extra ignore patterns."""
context.context_config = ContextConfig(ignore_patterns=["*.log"])
@given("a ContextConfig with hot_max_tokens {hot:d} and warm_max_decisions {warm:d}")
def step_context_config_memory_tiers(context: Context, hot: int, warm: int) -> None:
"""Create a ContextConfig with memory tier settings."""
context.context_config = ContextConfig(hot_max_tokens=hot, warm_max_decisions=warm)
@then("the nsproject ctx max_file_size should be {expected:d}")
def step_check_max_file_size(context: Context, expected: int) -> None:
"""Check max_file_size."""
assert context.context_config.max_file_size == expected
@then("the nsproject ctx max_total_size should be {expected:d}")
def step_check_max_total_size(context: Context, expected: int) -> None:
"""Check max_total_size."""
assert context.context_config.max_total_size == expected
@then('the nsproject ctx indexing_strategy should be "{expected}"')
def step_check_indexing_strategy(context: Context, expected: str) -> None:
"""Check indexing_strategy."""
assert context.context_config.indexing_strategy == expected
@then('the nsproject ctx chunking_policy should be "{expected}"')
def step_check_chunking_policy(context: Context, expected: str) -> None:
"""Check chunking_policy."""
assert context.context_config.chunking_policy == expected
@then("the nsproject ctx chunk_size should be {expected:d}")
def step_check_chunk_size(context: Context, expected: int) -> None:
"""Check chunk_size."""
assert context.context_config.chunk_size == expected
@then("the nsproject ctx summarize should be {expected}")
def step_check_summarize(context: Context, expected: str) -> None:
"""Check summarize."""
expected_bool = expected.lower() == "true"
assert context.context_config.summarize == expected_bool
@then("the nsproject ctx auto_refresh should be {expected}")
def step_check_auto_refresh(context: Context, expected: str) -> None:
"""Check auto_refresh."""
expected_bool = expected.lower() == "true"
assert context.context_config.auto_refresh == expected_bool
@then('the nsproject ctx temporal_scope should be "{expected}"')
def step_check_temporal_scope(context: Context, expected: str) -> None:
"""Check temporal_scope."""
assert context.context_config.temporal_scope == expected
@then("the nsproject ctx retention_policy should be empty")
def step_check_retention_policy_empty(context: Context) -> None:
"""Check retention_policy is None."""
assert context.context_config.retention_policy is None
@then("the nsproject ctx ignore patterns should include defaults")
def step_check_ignore_patterns_defaults(context: Context) -> None:
"""Check that default ignore patterns are present."""
from cleveragents.domain.models.core.project import (
DEFAULT_IGNORE_PATTERNS,
)
for pattern in DEFAULT_IGNORE_PATTERNS:
assert pattern in context.context_config.ignore_patterns, (
f"Default pattern '{pattern}' missing"
)
@then("the nsproject ctx ignore patterns should include the custom pattern")
def step_check_ignore_patterns_custom(context: Context) -> None:
"""Check that custom ignore pattern is present."""
assert "*.log" in context.context_config.ignore_patterns
@then("the nsproject ctx hot_max_tokens should be {expected:d}")
def step_check_hot_max_tokens(context: Context, expected: int) -> None:
"""Check hot_max_tokens."""
assert context.context_config.hot_max_tokens == expected
@then("the nsproject ctx warm_max_decisions should be {expected:d}")
def step_check_warm_max_decisions(context: Context, expected: int) -> None:
"""Check warm_max_decisions."""
assert context.context_config.warm_max_decisions == expected
@when("I try to mutate a ContextConfig field")
def step_try_mutate_context_config(context: Context) -> None:
"""Try to mutate a frozen ContextConfig."""
try:
context.context_config.chunk_size = 500 # type: ignore[misc]
context.project_error = None
except ValidationError as e:
context.project_error = e
@when("I try to create a ContextConfig with max_file_size {size:d}")
def step_try_create_context_config_bad_size(context: Context, size: int) -> None:
"""Try to create ContextConfig with invalid max_file_size."""
try:
ContextConfig(max_file_size=size)
context.project_error = None
except ValidationError as e:
context.project_error = e
@when("I try to create a ContextConfig with chunk_size {size:d}")
def step_try_create_context_config_bad_chunk(context: Context, size: int) -> None:
"""Try to create ContextConfig with invalid chunk_size."""
try:
ContextConfig(chunk_size=size)
context.project_error = None
except ValidationError as e:
context.project_error = e
# -- LinkedResource model ----------------------------------------------------
@given('a LinkedResource with resource_id "{rid}"')
def step_create_linked_resource(context: Context, rid: str) -> None:
"""Create a LinkedResource with defaults."""
context.linked_resource = LinkedResource(resource_id=rid)
@given('a LinkedResource with id "{rid}" and alias "{alias}"')
def step_create_linked_resource_with_alias(
context: Context, rid: str, alias: str
) -> None:
"""Create a LinkedResource with alias."""
context.linked_resource = LinkedResource(resource_id=rid, alias=alias)
@then('the nsproject linked rid should be "{expected}"')
def step_check_linked_rid(context: Context, expected: str) -> None:
"""Check linked resource_id."""
assert context.linked_resource.resource_id == expected
@then("the nsproject linked read_only should be {expected}")
def step_check_linked_readonly(context: Context, expected: str) -> None:
"""Check project_read_only."""
expected_bool = expected.lower() == "true"
assert context.linked_resource.project_read_only == expected_bool
@then("the nsproject linked alias should be empty")
def step_check_linked_alias_empty(context: Context) -> None:
"""Check alias is None."""
assert context.linked_resource.alias is None
@then('the nsproject linked alias should be "{expected}"')
def step_check_linked_alias(context: Context, expected: str) -> None:
"""Check alias."""
assert context.linked_resource.alias == expected
@then("the nsproject linked_at should be set")
def step_check_linked_at(context: Context) -> None:
"""Check linked_at is set."""
assert context.linked_resource.linked_at is not None
@when('I try to create a LinkedResource with resource_id "{rid}"')
def step_try_create_linked_resource_bad(context: Context, rid: str) -> None:
"""Try to create a LinkedResource with invalid resource_id."""
try:
LinkedResource(resource_id=rid)
context.project_error = None
except ValidationError as e:
context.project_error = e
@when("I try to create a LinkedResource with empty alias")
def step_try_create_linked_resource_empty_alias(
context: Context,
) -> None:
"""Try to create a LinkedResource with empty alias."""
try:
LinkedResource(resource_id=VALID_ULID, alias="")
context.project_error = None
except ValidationError as e:
context.project_error = e
@when("I try to mutate a LinkedResource field")
def step_try_mutate_linked_resource(context: Context) -> None:
"""Try to mutate a frozen LinkedResource."""
try:
context.linked_resource.alias = "x" # type: ignore[misc]
context.project_error = None
except ValidationError as e:
context.project_error = e
# -- NamespacedProject model -------------------------------------------------
@given('a NamespacedProject with name "{name}"')
def step_create_namespaced_project(context: Context, name: str) -> None:
"""Create a minimal NamespacedProject."""
context.project = NamespacedProject(name=name)
use_step_matcher("re") # type: ignore[attr-defined]
@given( # type: ignore[no-redef]
r'a project "(?P<name>[^"]+)" under namespace'
r' "(?P<namespace>[^"]+)"'
)
def step_create_namespaced_project_with_ns(
context: Context, name: str, namespace: str
) -> None:
"""Create a NamespacedProject with namespace."""
context.project = NamespacedProject(name=name, namespace=namespace)
@given( # type: ignore[no-redef]
r'a project "(?P<name>[^"]+)" under namespace'
r' "(?P<namespace>[^"]+)" on server "(?P<server>[^"]+)"'
)
def step_create_namespaced_project_full(
context: Context, name: str, namespace: str, server: str
) -> None:
"""Create a NamespacedProject with all identity fields."""
context.project = NamespacedProject(name=name, namespace=namespace, server=server)
use_step_matcher("parse") # type: ignore[attr-defined]
@then('the nsproject name should be "{expected}"')
def step_check_project_name(context: Context, expected: str) -> None:
"""Check project name."""
assert context.project.name == expected
@then('the nsproject namespace should be "{expected}"')
def step_check_project_namespace(context: Context, expected: str) -> None:
"""Check project namespace."""
assert context.project.namespace == expected
@then("the nsproject server should be empty")
def step_check_project_server_empty(context: Context) -> None:
"""Check project server is None."""
assert context.project.server is None
@then('the nsproject server should be "{expected}"')
def step_check_project_server(context: Context, expected: str) -> None:
"""Check project server."""
assert context.project.server == expected
@then("the nsproject linked resources should be empty")
def step_check_project_no_linked_resources(
context: Context,
) -> None:
"""Check linked_resources is empty."""
assert context.project.linked_resources == []
@then("the nsproject created_at should be set")
def step_check_project_created_at(context: Context) -> None:
"""Check created_at is set."""
assert context.project.created_at is not None
@then("the nsproject updated_at should be set")
def step_check_project_updated_at(context: Context) -> None:
"""Check updated_at is set."""
assert context.project.updated_at is not None
@then("the nsproject should be local")
def step_check_project_is_local(context: Context) -> None:
"""Check project is local."""
assert context.project.is_local, "Expected is_local=True"
@then("the nsproject should be remote")
def step_check_project_is_remote(context: Context) -> None:
"""Check project is remote."""
assert context.project.is_remote, "Expected is_remote=True"
# -- NamespacedProject validation -------------------------------------------
@when('I try to create a NamespacedProject with name "{name}"')
def step_try_create_project_bad_name(context: Context, name: str) -> None:
"""Try to create a NamespacedProject with invalid name."""
try:
NamespacedProject(name=name)
context.project_error = None
except ValidationError as e:
context.project_error = e
@when('I try to create a NamespacedProject with namespace "{namespace}"')
def step_try_create_project_bad_namespace(context: Context, namespace: str) -> None:
"""Try to create a NamespacedProject with reserved namespace."""
try:
NamespacedProject(name="test", namespace=namespace)
context.project_error = None
except ValidationError as e:
context.project_error = e
@when("I try to create a NamespacedProject with duplicate linked resources")
def step_try_create_project_dup_resources(
context: Context,
) -> None:
"""Try to create a NamespacedProject with duplicate links."""
rid = VALID_ULID
lr1 = LinkedResource(resource_id=rid)
lr2 = LinkedResource(resource_id=rid)
try:
NamespacedProject(name="test", linked_resources=[lr1, lr2])
context.project_error = None
except ValidationError as e:
context.project_error = e
# -- NamespacedProject domain methods ----------------------------------------
@when('I link resource "{rid}" to the project')
@given('I link resource "{rid}" to the project')
def step_link_resource(context: Context, rid: str) -> None:
"""Link a resource to the project."""
context.project = context.project.link_resource(rid)
@when('I link resource "{rid}" with read_only {ro} and alias "{alias}"')
@given('I link resource "{rid}" with read_only {ro} and alias "{alias}"')
def step_link_resource_with_options(
context: Context, rid: str, ro: str, alias: str
) -> None:
"""Link a resource with options."""
read_only = ro.lower() == "true"
context.project = context.project.link_resource(
rid, read_only=read_only, alias=alias
)
@when('I unlink resource "{rid}" from the project')
def step_unlink_resource(context: Context, rid: str) -> None:
"""Unlink a resource from the project."""
context.project = context.project.unlink_resource(rid)
@when('I try to link resource "{rid}" again')
def step_try_link_duplicate(context: Context, rid: str) -> None:
"""Try to link a resource that's already linked."""
try:
context.project.link_resource(rid)
context.project_error = None
except ValueError as e:
context.project_error = e
@when('I try to unlink resource "{rid}" from the project')
def step_try_unlink_nonexistent(context: Context, rid: str) -> None:
"""Try to unlink a resource that's not linked."""
try:
context.project.unlink_resource(rid)
context.project_error = None
except ValueError as e:
context.project_error = e
@when('I try to link resource "{rid}" to the project')
def step_try_link_resource(context: Context, rid: str) -> None:
"""Try to link a resource, might fail."""
try:
context.project = context.project.link_resource(rid)
context.project_error = None
except ValueError as e:
context.project_error = e
@when("I try to link an empty resource_id to the project")
def step_try_link_empty_resource(context: Context) -> None:
"""Try to link with empty resource_id."""
try:
context.project = context.project.link_resource("")
context.project_error = None
except ValueError as e:
context.project_error = e
@when("I try to unlink an empty resource_id from the project")
def step_try_unlink_empty_resource(context: Context) -> None:
"""Try to unlink with empty resource_id."""
try:
context.project = context.project.unlink_resource("")
context.project_error = None
except ValueError as e:
context.project_error = e
@when('I get linked resource "{rid}" from the project')
def step_get_linked_resource(context: Context, rid: str) -> None:
"""Get a linked resource by resource_id."""
context.retrieved_link = context.project.get_linked_resource(rid)
@when('I get linked resource by alias "{alias}" from the project')
def step_get_linked_resource_by_alias(context: Context, alias: str) -> None:
"""Get a linked resource by alias."""
context.retrieved_link = context.project.get_linked_resource_by_alias(alias)
@then("the nsproject should have {count:d} linked resource")
def step_check_linked_count(context: Context, count: int) -> None:
"""Check linked resource count."""
actual = len(context.project.linked_resources)
assert actual == count, f"Expected {count}, got {actual}"
@then('the nsproject first linked resource_id should be "{expected}"')
def step_check_linked_resource_id(context: Context, expected: str) -> None:
"""Check the first linked resource's resource_id."""
assert context.project.linked_resources[0].resource_id == expected
@then("the nsproject first linked resource should be read_only")
def step_check_first_linked_readonly(context: Context) -> None:
"""Check first linked resource is read_only."""
assert context.project.linked_resources[0].project_read_only
@then('the nsproject first linked resource alias should be "{expected}"')
def step_check_first_linked_alias(context: Context, expected: str) -> None:
"""Check first linked resource alias."""
assert context.project.linked_resources[0].alias == expected
@then("the nsproject retrieved link should not be empty")
def step_check_retrieved_link_not_none(context: Context) -> None:
"""Check retrieved link is not None."""
assert context.retrieved_link is not None
@then("the nsproject retrieved link should be empty")
def step_check_retrieved_link_none(context: Context) -> None:
"""Check retrieved link is None."""
assert context.retrieved_link is None
@then('the nsproject retrieved link alias should be "{expected}"')
def step_check_retrieved_link_alias(context: Context, expected: str) -> None:
"""Check retrieved link alias."""
assert context.retrieved_link.alias == expected
features/steps/resource_registry_model_steps.py
+482
View File
@@ -0,0 +1,482 @@
"""Step definitions for Resource Registry domain model tests."""
from __future__ import annotations
import time
from behave import given, then, when
from behave.runner import Context
from pydantic import ValidationError
from cleveragents.domain.models.core.resource import (
PhysVirt,
Resource,
ResourceCapabilities,
SandboxStrategy,
)
VALID_ULID = "01HZQX7K8M3N4P5R6S7T8V9W0X"
# -- PhysVirt enum -----------------------------------------------------------
@then('the PhysVirt enum should have values "{values}"')
def step_physvirt_values(context: Context, values: str) -> None:
"""Check PhysVirt enum has exactly the expected values."""
expected = {v.strip() for v in values.split(",")}
actual = {member.value for member in PhysVirt}
assert actual == expected, f"Expected {expected}, got {actual}"
@given('a PhysVirt value of "{value}"')
def step_create_physvirt(context: Context, value: str) -> None:
"""Create a PhysVirt enum value."""
context.physvirt = PhysVirt(value)
@then('the PhysVirt string value should be "{expected}"')
def step_check_physvirt_value(context: Context, expected: str) -> None:
"""Check PhysVirt string value."""
assert str(context.physvirt) == expected, (
f"Expected '{expected}', got '{context.physvirt}'"
)
@when('I try to create a PhysVirt with value "{value}"')
def step_try_create_physvirt(context: Context, value: str) -> None:
"""Try to create a PhysVirt enum value, expecting failure."""
try:
PhysVirt(value)
context.resource_error = None
except ValueError as e:
context.resource_error = e
@then("a resource validation error should be raised")
def step_check_resource_error(context: Context) -> None:
"""Check that a validation error was raised."""
assert context.resource_error is not None, (
"Expected a validation error but none was raised"
)
# -- SandboxStrategy enum ---------------------------------------------------
@then('the SandboxStrategy enum should have values "{values}"')
def step_sandbox_strategy_values(context: Context, values: str) -> None:
"""Check SandboxStrategy enum has exactly the expected values."""
expected = {v.strip() for v in values.split(",")}
actual = {member.value for member in SandboxStrategy}
assert actual == expected, f"Expected {expected}, got {actual}"
@given('a SandboxStrategy value of "{value}"')
def step_create_sandbox_strategy(context: Context, value: str) -> None:
"""Create a SandboxStrategy enum value."""
context.sandbox_strategy = SandboxStrategy(value)
@then('the SandboxStrategy string value should be "{expected}"')
def step_check_sandbox_strategy_value(context: Context, expected: str) -> None:
"""Check SandboxStrategy string value."""
assert str(context.sandbox_strategy) == expected, (
f"Expected '{expected}', got '{context.sandbox_strategy}'"
)
@when('I try to create a SandboxStrategy with value "{value}"')
def step_try_create_sandbox_strategy(context: Context, value: str) -> None:
"""Try to create a SandboxStrategy, expecting failure."""
try:
SandboxStrategy(value)
context.resource_error = None
except ValueError as e:
context.resource_error = e
# -- ResourceCapabilities ---------------------------------------------------
@given("default ResourceCapabilities")
def step_default_capabilities(context: Context) -> None:
"""Create default ResourceCapabilities."""
context.capabilities = ResourceCapabilities()
@given("ResourceCapabilities with readable false and checkpointable true")
def step_custom_capabilities(context: Context) -> None:
"""Create ResourceCapabilities with custom values."""
context.capabilities = ResourceCapabilities(readable=False, checkpointable=True)
@then("the capability {field} should be {expected}")
def step_check_capability(context: Context, field: str, expected: str) -> None:
"""Check a capability field value."""
actual = getattr(context.capabilities, field)
expected_bool = expected.lower() == "true"
assert actual == expected_bool, f"Expected {field}={expected_bool}, got {actual}"
@when("I try to mutate a ResourceCapabilities field")
def step_try_mutate_capabilities(context: Context) -> None:
"""Try to mutate a frozen ResourceCapabilities."""
try:
context.capabilities.readable = False # type: ignore[misc]
context.resource_error = None
except ValidationError as e:
context.resource_error = e
# -- Resource model - creation -----------------------------------------------
@given(
'a Resource with resource_id "{rid}" type "{rtype}"'
' and classification "{classification}"'
)
def step_create_minimal_resource(
context: Context, rid: str, rtype: str, classification: str
) -> None:
"""Create a minimal Resource."""
context.resource = Resource(
resource_id=rid,
resource_type_name=rtype,
classification=PhysVirt(classification),
)
@given(
'a full Resource with id "{rid}" name "{name}" type "{rtype}"'
' classification "{classification}" description "{desc}"'
)
def step_create_full_resource(
context: Context,
rid: str,
name: str,
rtype: str,
classification: str,
desc: str,
) -> None:
"""Create a full Resource with all fields."""
context.resource = Resource(
resource_id=rid,
name=name,
resource_type_name=rtype,
classification=PhysVirt(classification),
description=desc,
)
@given('a Resource with sandbox strategy "{strategy}"')
def step_create_resource_with_sandbox(context: Context, strategy: str) -> None:
"""Create a Resource with a sandbox strategy override."""
context.resource = Resource(
resource_id=VALID_ULID,
resource_type_name="git-checkout",
classification=PhysVirt.PHYSICAL,
sandbox_strategy=SandboxStrategy(strategy),
)
@given("a Resource with sandboxable false and checkpointable true")
def step_create_resource_with_capabilities(context: Context) -> None:
"""Create a Resource with custom capabilities."""
context.resource = Resource(
resource_id=VALID_ULID,
resource_type_name="git-checkout",
classification=PhysVirt.PHYSICAL,
capabilities=ResourceCapabilities(sandboxable=False, checkpointable=True),
)
@given("a Resource with writable false")
def step_create_readonly_resource(context: Context) -> None:
"""Create a Resource with writable=false."""
context.resource = Resource(
resource_id=VALID_ULID,
resource_type_name="git-checkout",
classification=PhysVirt.PHYSICAL,
capabilities=ResourceCapabilities(writable=False),
)
@given('a Resource with content_hash "{hash_val}"')
def step_create_resource_with_hash(context: Context, hash_val: str) -> None:
"""Create a Resource with content_hash."""
context.resource = Resource(
resource_id=VALID_ULID,
resource_type_name="git-checkout",
classification=PhysVirt.PHYSICAL,
content_hash=hash_val,
)
@given('a Resource with location "{location}"')
def step_create_resource_with_location(context: Context, location: str) -> None:
"""Create a Resource with location."""
context.resource = Resource(
resource_id=VALID_ULID,
resource_type_name="git-checkout",
classification=PhysVirt.PHYSICAL,
location=location,
)
@given("a Resource with parents and children")
def step_create_resource_with_dag(context: Context) -> None:
"""Create a Resource with parent/child references."""
parent_id = "01HZQX7K8M3N4P5R6S7T8V9W01"
child1_id = "01HZQX7K8M3N4P5R6S7T8V9W02"
child2_id = "01HZQX7K8M3N4P5R6S7T8V9W03"
context.resource = Resource(
resource_id=VALID_ULID,
resource_type_name="git-checkout",
classification=PhysVirt.PHYSICAL,
parents=[parent_id],
children=[child1_id, child2_id],
)
# -- Resource model - assertion steps ----------------------------------------
@then('the resource resource_id should be "{expected}"')
def step_check_resource_id(context: Context, expected: str) -> None:
"""Check resource_id."""
assert context.resource.resource_id == expected
@then('the resource type name should be "{expected}"')
def step_check_resource_type_name(context: Context, expected: str) -> None:
"""Check resource_type_name."""
assert context.resource.resource_type_name == expected
@then('the resource classification should be "{expected}"')
def step_check_classification(context: Context, expected: str) -> None:
"""Check classification."""
assert context.resource.classification == expected
@then("the resource name should be empty")
def step_check_name_empty(context: Context) -> None:
"""Check name is None."""
assert context.resource.name is None
@then('the resource name should be "{expected}"')
def step_check_resource_name(context: Context, expected: str) -> None:
"""Check resource name."""
assert context.resource.name == expected
@then("the resource description should be empty")
def step_check_description_empty(context: Context) -> None:
"""Check description is None."""
assert context.resource.description is None
@then('the resource description should be "{expected}"')
def step_check_resource_description(context: Context, expected: str) -> None:
"""Check description."""
assert context.resource.description == expected
@then("the resource properties should be empty")
def step_check_properties_empty(context: Context) -> None:
"""Check properties dict is empty."""
assert context.resource.properties == {}
@then("the resource parents should be empty")
def step_check_parents_empty(context: Context) -> None:
"""Check parents list is empty."""
assert context.resource.parents == []
@then("the resource children should be empty")
def step_check_children_empty(context: Context) -> None:
"""Check children list is empty."""
assert context.resource.children == []
@then("the resource linked_projects should be empty")
def step_check_linked_projects_empty(context: Context) -> None:
"""Check linked_projects list is empty."""
assert context.resource.linked_projects == []
@then("the resource created_at should be set")
def step_check_created_at(context: Context) -> None:
"""Check created_at is not None."""
assert context.resource.created_at is not None
@then("the resource updated_at should be set")
def step_check_updated_at(context: Context) -> None:
"""Check updated_at is not None."""
assert context.resource.updated_at is not None
@then('the resource sandbox strategy should be "{expected}"')
def step_check_sandbox_strategy(context: Context, expected: str) -> None:
"""Check sandbox_strategy."""
assert context.resource.sandbox_strategy == expected
@then("the resource can_sandbox should be {expected}")
def step_check_can_sandbox(context: Context, expected: str) -> None:
"""Check can_sandbox property."""
expected_bool = expected.lower() == "true"
assert context.resource.can_sandbox == expected_bool
@then("the resource can_checkpoint should be {expected}")
def step_check_can_checkpoint(context: Context, expected: str) -> None:
"""Check can_checkpoint property."""
expected_bool = expected.lower() == "true"
assert context.resource.can_checkpoint == expected_bool
@then("the resource is_physical should be {expected}")
def step_check_is_physical(context: Context, expected: str) -> None:
"""Check is_physical property."""
expected_bool = expected.lower() == "true"
assert context.resource.is_physical == expected_bool
@then("the resource is_virtual should be {expected}")
def step_check_is_virtual(context: Context, expected: str) -> None:
"""Check is_virtual property."""
expected_bool = expected.lower() == "true"
assert context.resource.is_virtual == expected_bool
@then("the resource is_read_only should be {expected}")
def step_check_is_read_only(context: Context, expected: str) -> None:
"""Check is_read_only property."""
expected_bool = expected.lower() == "true"
assert context.resource.is_read_only == expected_bool
@then('the resource content_hash should be "{expected}"')
def step_check_content_hash(context: Context, expected: str) -> None:
"""Check content_hash."""
assert context.resource.content_hash == expected
@then('the resource location should be "{expected}"')
def step_check_location(context: Context, expected: str) -> None:
"""Check location."""
assert context.resource.location == expected
@then("the resource should have {count:d} parent")
def step_check_parent_count(context: Context, count: int) -> None:
"""Check parent count."""
assert len(context.resource.parents) == count
@then("the resource should have {count:d} children")
def step_check_child_count(context: Context, count: int) -> None:
"""Check children count."""
assert len(context.resource.children) == count
# -- Resource model - validation failures ------------------------------------
@when('I try to create a Resource with resource_id "{rid}"')
def step_try_create_resource_bad_ulid(context: Context, rid: str) -> None:
"""Try to create a Resource with invalid ULID."""
try:
Resource(
resource_id=rid,
resource_type_name="git-checkout",
classification=PhysVirt.PHYSICAL,
)
context.resource_error = None
except ValidationError as e:
context.resource_error = e
@when("I try to create a Resource with empty resource_type_name")
def step_try_create_resource_empty_type(context: Context) -> None:
"""Try to create a Resource with empty type name."""
try:
Resource(
resource_id=VALID_ULID,
resource_type_name="",
classification=PhysVirt.PHYSICAL,
)
context.resource_error = None
except ValidationError as e:
context.resource_error = e
@when("I try to create a Resource with empty string name")
def step_try_create_resource_empty_name(context: Context) -> None:
"""Try to create a Resource with empty string name."""
try:
Resource(
resource_id=VALID_ULID,
name="",
resource_type_name="git-checkout",
classification=PhysVirt.PHYSICAL,
)
context.resource_error = None
except ValidationError as e:
context.resource_error = e
@when("I try to create a Resource with empty location")
def step_try_create_resource_empty_location(context: Context) -> None:
"""Try to create a Resource with empty location."""
try:
Resource(
resource_id=VALID_ULID,
resource_type_name="git-checkout",
classification=PhysVirt.PHYSICAL,
location="",
)
context.resource_error = None
except ValidationError as e:
context.resource_error = e
# -- Resource model - frozen -------------------------------------------------
@when("I try to mutate a Resource field")
def step_try_mutate_resource(context: Context) -> None:
"""Try to mutate a frozen Resource."""
try:
context.resource.name = "new-name" # type: ignore[misc]
context.resource_error = None
except ValidationError as e:
context.resource_error = e
# -- Resource model - with_updated_at ---------------------------------------
@when("I call with_updated_at on the resource")
def step_call_with_updated_at(context: Context) -> None:
"""Call with_updated_at and store old/new resources."""
context.old_resource = context.resource
time.sleep(0.01) # Ensure timestamp differs
context.new_resource = context.resource.with_updated_at()
@then("the new resource should have a different updated_at")
def step_check_different_updated_at(context: Context) -> None:
"""Check updated_at changed."""
assert context.new_resource.updated_at != context.old_resource.updated_at
@then("the new resource should have the same resource_id")
def step_check_same_resource_id(context: Context) -> None:
"""Check resource_id unchanged."""
assert context.new_resource.resource_id == context.old_resource.resource_id
src/cleveragents/domain/models/core/__init__.py
+24
View File
@@ -50,10 +50,24 @@ from cleveragents.domain.models.core.plan_legacy import (
PlanStatus,
)
from cleveragents.domain.models.core.project import (
ContextConfig,
LinkedResource,
NamespacedProject,
ParsedName,
TemporalScope,
parse_namespaced_name,
)
from cleveragents.domain.models.core.project_legacy import (
Project,
ProjectSettings,
ProjectStats,
)
from cleveragents.domain.models.core.resource import (
PhysVirt,
Resource,
ResourceCapabilities,
SandboxStrategy,
)
__all__ = [
"ActionState",
@@ -62,6 +76,7 @@ __all__ = [
"ChangeSet",
"CloudBillingFields",
"Context",
"ContextConfig",
"ContextFile",
"ContextType",
"ContextUpdateResult",
@@ -72,13 +87,17 @@ __all__ = [
"InvariantSource",
"Invite",
"LifecyclePlan",
"LinkedResource",
"MaxContextCount",
"NamespacedName",
"NamespacedProject",
"Operation",
"OperationType",
"Org",
"OrgRole",
"OrgUser",
"ParsedName",
"PhysVirt",
"Plan",
"PlanBuild",
"PlanIdentity",
@@ -92,7 +111,12 @@ __all__ = [
"ProjectLink",
"ProjectSettings",
"ProjectStats",
"Resource",
"ResourceCapabilities",
"SandboxStrategy",
"SummaryForUpdateContextParams",
"TemporalScope",
"User",
"can_transition",
"parse_namespaced_name",
]
src/cleveragents/domain/models/core/project.py
+515 -66
View File
@@ -1,85 +1,534 @@
"""Project domain model for CleverAgents.
Based on Phase 0 discovery and ADR-004 (Pydantic Validation).
Contains both the legacy Project model (for backward compatibility with
existing services) and the spec-aligned models:
- LinkedResource (project-resource link record)
- ContextConfig (project-level context configuration)
- NamespacedProject (spec-aligned project identified solely by namespaced name)
Spec references:
- Project Data Model (lines 6477-6511)
- Namespaces (lines 6524-6584)
- project_resources DDL (lines 26726-26733)
- Context configuration CLI flags (lines 94-114)
- ADR-004: Pydantic v2 Validation
"""
from datetime import datetime
from pathlib import Path
from __future__ import annotations
from pydantic import BaseModel, ConfigDict, Field, field_validator
import re
from datetime import UTC, datetime
from enum import StrEnum
from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator
# ---------------------------------------------------------------------------
# Namespace parsing utilities
# ---------------------------------------------------------------------------
# Full namespaced name pattern: [[server:]namespace/]name
# Examples: "local/my-project", "dev:freemo/api-service",
# "my-project" (defaults to local/)
_SERVER_NS_NAME_RE = re.compile(
r"^(?:(?P<server>[a-zA-Z][a-zA-Z0-9_-]*):)?"
r"(?P<namespace>[a-zA-Z][a-zA-Z0-9_-]*)/"
r"(?P<name>[a-zA-Z][a-zA-Z0-9_-]*)$"
)
# Name-only pattern (no namespace, no server)
_BARE_NAME_RE = re.compile(r"^[a-zA-Z][a-zA-Z0-9_-]*$")
RESERVED_NAMESPACES = frozenset(
{
"system",
"internal",
"admin",
"root",
}
)
# Built-in provider namespaces (reserved for LLM actors, cannot be used for projects)
PROVIDER_NAMESPACES = frozenset(
{
"openai",
"anthropic",
"google",
"gemini",
"deepseek",
"mistral",
"perplexity",
"qwen",
"amazon",
}
)
DEFAULT_NAMESPACE = "local"
class ProjectSettings(BaseModel):
"""Project-specific settings and configuration."""
class ParsedName(BaseModel):
"""Result of parsing a [[server:]namespace/]name string.
auto_build: bool = Field(False, description="Automatically build plans")
auto_apply: bool = Field(False, description="Automatically apply changes")
confirm_apply: bool = Field(
True, description="Require confirmation before applying"
)
max_context_size: int = Field(
52428800, description="Maximum context size in bytes (50MB)"
)
default_model: str = Field("mock-gpt", description="Default AI model")
include_paths: list[str] = Field(
default_factory=list, description="Relative include globs"
)
exclude_paths: list[str] = Field(
default_factory=list, description="Relative exclude globs"
)
model_config = ConfigDict(
str_strip_whitespace=True,
validate_assignment=True,
)
class ProjectStats(BaseModel):
"""Statistics for a project."""
plans: int = Field(0, ge=0)
context_files: int = Field(0, ge=0)
changes: int = Field(0, ge=0)
current_plan: str | None = Field(None)
model_config = ConfigDict(
str_strip_whitespace=True,
validate_assignment=True,
)
class Project(BaseModel):
"""Domain model for a CleverAgents project.
A project represents a workspace with plans, contexts, and changes.
Spec reference: Namespace format (line 47 glossary, lines 6524-6584).
"""
id: int | None = Field(None, description="Project ID")
name: str = Field(..., min_length=1, max_length=255)
path: Path = Field(..., description="Project root path")
created_at: datetime = Field(default_factory=datetime.now)
updated_at: datetime = Field(default_factory=datetime.now)
settings: ProjectSettings = Field(default_factory=lambda: ProjectSettings()) # type: ignore
current_plan_id: int | None = Field(None)
server: str | None = Field(
default=None, description="Server prefix (e.g., 'dev', 'prod')"
)
namespace: str = Field(
default=DEFAULT_NAMESPACE,
description="Namespace (e.g., 'local', 'freemo', 'cleverthis')",
)
name: str = Field(..., min_length=1, description="Entity name")
model_config = ConfigDict(frozen=True)
@property
def qualified_name(self) -> str:
"""Return the fully qualified name: [[server:]namespace/]name."""
parts: list[str] = []
if self.server is not None:
parts.append(f"{self.server}:")
parts.append(f"{self.namespace}/{self.name}")
return "".join(parts)
@property
def namespaced_name(self) -> str:
"""Return namespace/name (without server prefix)."""
return f"{self.namespace}/{self.name}"
@property
def is_local(self) -> bool:
"""Check if this is a local-only entity."""
return self.namespace == DEFAULT_NAMESPACE and self.server is None
@property
def is_remote(self) -> bool:
"""Check if this entity lives on a server.
True when namespace is non-local or server-qualified.
"""
return self.server is not None or self.namespace != DEFAULT_NAMESPACE
def parse_namespaced_name(value: str) -> ParsedName:
"""Parse a [[server:]namespace/]name string.
Spec reference: Namespace format (line 47, lines 6524-6584).
Args:
value: The name string to parse.
Returns:
ParsedName with server, namespace, and name components.
Raises:
ValueError: If the string is empty, has invalid characters,
or uses a reserved/provider namespace.
"""
if not value or not value.strip():
raise ValueError("Name cannot be empty")
value = value.strip()
# Try full pattern: [[server:]namespace/]name
match = _SERVER_NS_NAME_RE.match(value)
if match:
server = match.group("server")
namespace = match.group("namespace")
name = match.group("name")
if namespace in RESERVED_NAMESPACES:
raise ValueError(
f"Namespace '{namespace}' is reserved. "
f"Reserved: {', '.join(sorted(RESERVED_NAMESPACES))}"
)
if namespace in PROVIDER_NAMESPACES:
raise ValueError(
f"Namespace '{namespace}' is reserved for built-in LLM actors "
f"and cannot be used for projects"
)
return ParsedName(server=server, namespace=namespace, name=name)
# Try bare name (no namespace): defaults to local/
if _BARE_NAME_RE.match(value):
return ParsedName(server=None, namespace=DEFAULT_NAMESPACE, name=value)
raise ValueError(
f"Invalid name format: '{value}'. "
f"Expected [[server:]namespace/]name where each part starts with "
f"a letter and contains only letters, digits, hyphens, and underscores."
)
# ---------------------------------------------------------------------------
# Spec-aligned models
# ---------------------------------------------------------------------------
DEFAULT_IGNORE_PATTERNS: list[str] = [
".git/",
"node_modules/",
"__pycache__/",
".venv/",
"*.pyc",
".DS_Store",
]
class TemporalScope(StrEnum):
"""Temporal scope for context retrieval."""
CURRENT = "current"
RECENT = "recent"
ALL = "all"
class ContextConfig(BaseModel):
"""Project-level context configuration.
Spec reference: Project context configuration (lines 6503-6511)
and ``agents project context set`` CLI flags (lines 94-114).
"""
# Path filtering
ignore_patterns: list[str] = Field(
default_factory=lambda: list(DEFAULT_IGNORE_PATTERNS),
description="File patterns to ignore during indexing (.gitignore semantics)",
)
include_patterns: list[str] = Field(
default_factory=list,
description="File patterns to include (empty means include all)",
)
# Size limits
max_file_size: int = Field(
default=1_000_000,
gt=0,
description="Maximum file size in bytes",
)
max_total_size: int = Field(
default=52_428_800,
gt=0,
description="Maximum total context size in bytes (50MB default)",
)
# Indexing and chunking
indexing_strategy: str = Field(
default="full_text",
description="Indexing strategy (e.g., 'full_text', 'semantic')",
)
chunking_policy: str = Field(
default="smart",
description="Chunking policy (e.g., 'smart', 'fixed')",
)
chunk_size: int = Field(
default=1000,
gt=0,
description="Chunk size in tokens",
)
# Memory tier configuration
hot_max_tokens: int | None = Field(
default=None, description="Max tokens for hot tier context"
)
warm_max_decisions: int | None = Field(
default=None, description="Max decisions for warm tier"
)
cold_max_decisions: int | None = Field(
default=None, description="Max decisions for cold tier"
)
# Summarization
summarize: bool = Field(default=True, description="Whether to enable summarization")
summary_max_tokens: int | None = Field(
default=None, description="Max tokens for summaries"
)
# Temporal and refresh
temporal_scope: TemporalScope = Field(
default=TemporalScope.CURRENT,
description="Temporal scope for context retrieval",
)
auto_refresh: bool = Field(
default=True, description="Auto-refresh context on changes"
)
# Retention policy (spec: "context retention policy")
retention_policy: str | None = Field(
default=None,
description="Context retention policy",
)
model_config = ConfigDict(
frozen=True,
str_strip_whitespace=True,
use_enum_values=True,
)
@field_validator("ignore_patterns")
@classmethod
def merge_default_ignore_patterns(
cls: type[ContextConfig], v: list[str]
) -> list[str]:
"""Ensure default ignore patterns are always present."""
merged = list(DEFAULT_IGNORE_PATTERNS)
for pattern in v:
if pattern not in merged:
merged.append(pattern)
return merged
class LinkedResource(BaseModel):
"""A project-resource link record.
Spec reference: Project Data Model, section 2 "Linked Resources"
(lines 6488-6502) and project_resources DDL (lines 26726-26733).
Projects link to resources from the Resource Registry rather than
embedding them. A resource can be linked to multiple projects.
"""
resource_id: str = Field(
...,
description="ULID reference to a Resource Registry entry",
pattern=r"^[0-9A-HJKMNP-TV-Z]{26}$",
)
project_read_only: bool = Field(
default=False,
description="Project-level read-only override",
)
alias: str | None = Field(
default=None,
description="Optional short name for referencing within the project",
)
linked_at: datetime = Field(
default_factory=lambda: datetime.now(tz=UTC),
description="When this resource was linked to the project",
)
model_config = ConfigDict(frozen=True)
@field_validator("alias")
@classmethod
def validate_alias(cls: type[LinkedResource], v: str | None) -> str | None:
"""Validate alias is non-empty if provided."""
if v is not None and not v.strip():
raise ValueError("Alias cannot be empty (use None instead)")
return v
class NamespacedProject(BaseModel):
"""Spec-aligned project model.
Spec reference: Project Data Model (lines 6477-6511).
A project is identified SOLELY by its namespaced name -- no ULID
or integer ID is generated. The namespacing scheme
([[server:]namespace/]name) ensures global uniqueness.
"""
# Identity -- namespaced name IS the identifier (no ULID)
name: str = Field(
...,
min_length=1,
max_length=255,
description="Project name (without namespace prefix)",
)
namespace: str = Field(
default=DEFAULT_NAMESPACE,
description="Project namespace (e.g., 'local', 'freemo', 'cleverthis')",
)
server: str | None = Field(
default=None,
description="Server prefix for multi-server disambiguation",
)
# Description
description: str | None = Field(
default=None, description="Human-readable project description"
)
# Linked resources (spec section 2)
linked_resources: list[LinkedResource] = Field(
default_factory=list,
description="Resources linked to this project from the Resource Registry",
)
# Context configuration (spec section 3)
context_config: ContextConfig = Field(
default_factory=ContextConfig,
description="Project-level context configuration",
)
# Timestamps
created_at: datetime = Field(
default_factory=lambda: datetime.now(tz=UTC),
)
updated_at: datetime = Field(
default_factory=lambda: datetime.now(tz=UTC),
)
model_config = ConfigDict(
str_strip_whitespace=True,
validate_assignment=True,
)
# -- Validators ----------------------------------------------------------
@field_validator("name")
@classmethod
def validate_name(cls: type["Project"], v: str) -> str:
"""Validate project name."""
if not v.replace("-", "").replace("_", "").replace(" ", "").isalnum():
def validate_name(cls: type[NamespacedProject], v: str) -> str:
"""Validate project name format.
Must start with a letter, alphanumeric + hyphens + underscores.
"""
if not _BARE_NAME_RE.match(v):
raise ValueError(
"Name must be alphanumeric with hyphens, underscores, or spaces"
f"Project name '{v}' is invalid. Must start with a letter "
f"and contain only letters, digits, hyphens, and underscores."
)
return v
@field_validator("path")
@field_validator("namespace")
@classmethod
def validate_path(cls: type["Project"], v: Path) -> Path:
"""Ensure path is absolute."""
return v.resolve()
def validate_namespace(cls: type[NamespacedProject], v: str) -> str:
"""Validate namespace: must be a valid namespace, not reserved."""
if not _BARE_NAME_RE.match(v):
raise ValueError(
f"Namespace '{v}' is invalid. Must start with a letter "
f"and contain only letters, digits, hyphens, and underscores."
)
if v in RESERVED_NAMESPACES:
raise ValueError(
f"Namespace '{v}' is reserved. "
f"Reserved: {', '.join(sorted(RESERVED_NAMESPACES))}"
)
if v in PROVIDER_NAMESPACES:
raise ValueError(f"Namespace '{v}' is reserved for built-in LLM actors.")
return v
model_config = ConfigDict(
str_strip_whitespace=True,
validate_assignment=True,
arbitrary_types_allowed=True, # Allow Path
)
@model_validator(mode="after")
def validate_linked_resources_unique(self) -> NamespacedProject:
"""Ensure no duplicate resource_ids in linked_resources."""
ids = [lr.resource_id for lr in self.linked_resources]
if len(ids) != len(set(ids)):
raise ValueError("Duplicate resource_id in linked_resources")
return self
# -- Properties ----------------------------------------------------------
@property
def namespaced_name(self) -> str:
"""Return namespace/name."""
return f"{self.namespace}/{self.name}"
@property
def qualified_name(self) -> str:
"""Return [[server:]namespace/]name."""
if self.server is not None:
return f"{self.server}:{self.namespace}/{self.name}"
return self.namespaced_name
@property
def is_local(self) -> bool:
"""Check if this is a local-only project."""
return self.namespace == DEFAULT_NAMESPACE and self.server is None
@property
def is_remote(self) -> bool:
"""Check if this project lives on a server."""
return self.server is not None or self.namespace != DEFAULT_NAMESPACE
# -- Domain Methods ------------------------------------------------------
def link_resource(
self,
resource_id: str,
read_only: bool = False,
alias: str | None = None,
) -> NamespacedProject:
"""Return a new project with the resource linked.
Args:
resource_id: ULID of the resource to link.
read_only: Whether this resource is read-only in this project context.
alias: Optional short name for the resource within the project.
Raises:
ValueError: If resource_id is empty or already linked.
"""
if not resource_id or not resource_id.strip():
raise ValueError("resource_id cannot be empty")
if any(lr.resource_id == resource_id for lr in self.linked_resources):
raise ValueError(
f"Resource '{resource_id}' is already linked to this project"
)
link = LinkedResource(
resource_id=resource_id,
project_read_only=read_only,
alias=alias,
)
new_resources = [*self.linked_resources, link]
return self.model_copy(
update={
"linked_resources": new_resources,
"updated_at": datetime.now(tz=UTC),
}
)
def unlink_resource(self, resource_id: str) -> NamespacedProject:
"""Return a new project with the resource unlinked.
Args:
resource_id: ULID of the resource to unlink.
Raises:
ValueError: If resource_id is empty or not linked.
"""
if not resource_id or not resource_id.strip():
raise ValueError("resource_id cannot be empty")
if not any(lr.resource_id == resource_id for lr in self.linked_resources):
raise ValueError(f"Resource '{resource_id}' is not linked to this project")
new_resources = [
lr for lr in self.linked_resources if lr.resource_id != resource_id
]
return self.model_copy(
update={
"linked_resources": new_resources,
"updated_at": datetime.now(tz=UTC),
}
)
def get_linked_resource(self, resource_id: str) -> LinkedResource | None:
"""Get a linked resource by resource_id.
Args:
resource_id: ULID to look up.
Raises:
ValueError: If resource_id is empty.
"""
if not resource_id or not resource_id.strip():
raise ValueError("resource_id cannot be empty")
for lr in self.linked_resources:
if lr.resource_id == resource_id:
return lr
return None
def get_linked_resource_by_alias(self, alias: str) -> LinkedResource | None:
"""Get a linked resource by alias.
Args:
alias: Alias to look up.
Raises:
ValueError: If alias is empty.
"""
if not alias or not alias.strip():
raise ValueError("alias cannot be empty")
for lr in self.linked_resources:
if lr.alias == alias:
return lr
return None
src/cleveragents/domain/models/core/project_legacy.py
+99
View File
@@ -0,0 +1,99 @@
"""Legacy project models (preserved for backward compatibility).
New code should use the spec-aligned models in project.py:
- NamespacedProject, LinkedResource, ContextConfig, ParsedName
These legacy models exist to avoid breaking existing services,
repositories, and CLI commands that depend on the old API.
"""
from __future__ import annotations
from datetime import datetime
from pathlib import Path
from pydantic import BaseModel, ConfigDict, Field, field_validator
class ProjectSettings(BaseModel):
"""Project-specific settings and configuration (legacy)."""
auto_build: bool = Field(default=False, description="Automatically build plans")
auto_apply: bool = Field(
default=False,
description="Automatically apply changes",
)
confirm_apply: bool = Field(
default=True,
description="Require confirmation before applying",
)
max_context_size: int = Field(
default=52428800,
description="Maximum context size in bytes (50MB)",
)
default_model: str = Field(default="mock-gpt", description="Default AI model")
include_paths: list[str] = Field(
default_factory=list, description="Relative include globs"
)
exclude_paths: list[str] = Field(
default_factory=list, description="Relative exclude globs"
)
model_config = ConfigDict(
str_strip_whitespace=True,
validate_assignment=True,
)
class ProjectStats(BaseModel):
"""Statistics for a project (legacy)."""
plans: int = Field(default=0, ge=0)
context_files: int = Field(default=0, ge=0)
changes: int = Field(default=0, ge=0)
current_plan: str | None = Field(default=None)
model_config = ConfigDict(
str_strip_whitespace=True,
validate_assignment=True,
)
class Project(BaseModel):
"""Legacy project model (preserved for backward compatibility).
New code should use NamespacedProject instead. This model exists
to avoid breaking existing services, repositories, and CLI commands.
"""
id: int | None = Field(default=None, description="Project ID")
name: str = Field(..., min_length=1, max_length=255)
path: Path = Field(..., description="Project root path")
created_at: datetime = Field(default_factory=datetime.now)
updated_at: datetime = Field(default_factory=datetime.now)
settings: ProjectSettings = Field(
default_factory=lambda: ProjectSettings() # type: ignore[arg-type]
)
current_plan_id: int | None = Field(default=None)
@field_validator("name")
@classmethod
def validate_name(cls: type[Project], v: str) -> str:
"""Validate project name."""
if not v.replace("-", "").replace("_", "").replace(" ", "").isalnum():
raise ValueError(
"Name must be alphanumeric with hyphens, underscores, or spaces"
)
return v
@field_validator("path")
@classmethod
def validate_path(cls: type[Project], v: Path) -> Path:
"""Ensure path is absolute."""
return v.resolve()
model_config = ConfigDict(
str_strip_whitespace=True,
validate_assignment=True,
arbitrary_types_allowed=True,
)
src/cleveragents/domain/models/core/resource.py
+250
View File
@@ -0,0 +1,250 @@
"""Resource domain models for CleverAgents.
Implements the Resource Registry data model as defined in docs/specification.md:
- PhysVirt enum (physical | virtual classification)
- SandboxStrategy enum (git_worktree | copy_on_write |
transaction_rollback | snapshot | none)
- ResourceCapabilities (readable, writable, sandboxable, checkpointable)
- ResourceRecord (the core Resource domain model)
Based on:
- Specification: ResourceRecord UML (lines 9932-9944)
- Specification: PhysVirt enum (lines 9961-9964)
- Specification: SandboxStrategy JSON schema (lines 17135-17138)
- Specification: Resource Capabilities (lines 9892-9903)
- Specification: resources DDL (lines 26696-26713)
- ADR-004: Pydantic v2 Validation
"""
from __future__ import annotations
import re
from datetime import UTC, datetime
from enum import StrEnum
from pydantic import BaseModel, ConfigDict, Field, field_validator
# ULID: 26 characters, Crockford's base32
ULID_PATTERN = r"^[0-9A-HJKMNP-TV-Z]{26}$"
# Namespaced name: [[server:]namespace/]name
# Built-in resource types are unnamespaced (e.g., "git-checkout").
# Custom types follow: namespace/name (e.g., "local/my-type").
NAMESPACED_NAME_PATTERN = re.compile(
r"^(?:(?P<server>[a-zA-Z0-9_-]+):)?(?P<namespace>[a-zA-Z0-9_-]+)/(?P<name>[a-zA-Z0-9_-]+)$"
)
class PhysVirt(StrEnum):
"""Classification of a resource as physical or virtual.
Spec reference: PhysVirt enum (lines 9961-9964).
- physical: Represents a real, tangible resource (file, directory, git repo).
- virtual: Represents a derived/computed resource (git branch, tree entry).
"""
PHYSICAL = "physical"
VIRTUAL = "virtual"
class SandboxStrategy(StrEnum):
"""Sandbox isolation strategy for a resource.
Spec reference: JSON schema enum (lines 17135-17138).
Exactly 5 values as defined in the specification.
"""
GIT_WORKTREE = "git_worktree"
COPY_ON_WRITE = "copy_on_write"
TRANSACTION_ROLLBACK = "transaction_rollback"
SNAPSHOT = "snapshot"
NONE = "none"
class ResourceCapabilities(BaseModel):
"""Capabilities declared by a resource, derived from its resource type.
Spec reference: Resource Capabilities (lines 9892-9903).
These are used by the tool execution flow to validate that a tool's
resource binding is compatible with the resource.
"""
readable: bool = Field(default=True, description="Whether the resource can be read")
writable: bool = Field(
default=True, description="Whether the resource can be modified"
)
sandboxable: bool = Field(
default=True,
description="Whether the resource supports sandbox isolation",
)
checkpointable: bool = Field(
default=False,
description="Whether the resource supports checkpoint/rollback",
)
model_config = ConfigDict(frozen=True)
class Resource(BaseModel):
"""Domain model for a Resource Registry entry (ResourceRecord).
Spec reference: ResourceRecord UML (lines 9932-9944) and
resources DDL (lines 26696-26713).
Resources are identified by ULID (resource_id). They may optionally
have a namespaced name (NULL for auto-discovered children).
"""
# Identity -- resource_id is the primary key (ULID)
resource_id: str = Field(
...,
description="Unique ULID identifier (primary key)",
pattern=ULID_PATTERN,
)
name: str | None = Field(
default=None,
description="Namespaced name (NULL for auto-discovered children)",
)
# Type classification
resource_type_name: str = Field(
...,
min_length=1,
description="Namespaced type name (e.g., 'git-checkout', 'local/my-type')",
)
classification: PhysVirt = Field(
...,
description="Physical or virtual classification",
)
# Descriptive
description: str | None = Field(
default=None,
description="Human-readable description",
)
# Type-specific properties (JSON in DB)
properties: dict[str, str | int | float | bool | None] = Field(
default_factory=dict,
description="Type-specific properties (stored as JSON)",
)
# Location and hashing
location: str | None = Field(
default=None,
description="Physical location (physical resources only)",
)
content_hash: str | None = Field(
default=None,
description="Content hash for equivalence tracking",
)
# Sandbox strategy override (per-resource, may differ from type default)
sandbox_strategy: SandboxStrategy | None = Field(
default=None,
description="Sandbox strategy override per resource",
)
# Capabilities (derived from resource type)
capabilities: ResourceCapabilities = Field(
default_factory=ResourceCapabilities,
description=(
"Resource capabilities (readable, writable, sandboxable, checkpointable)"
),
)
# DAG relationships (loaded lazily in practice,
# represented here for domain completeness)
parents: list[str] = Field(
default_factory=list,
description="Parent resource_id list (DAG upward references)",
)
children: list[str] = Field(
default_factory=list,
description="Child resource_id list (DAG downward references)",
)
# Cross-references
linked_projects: list[str] = Field(
default_factory=list,
description="Project names this resource is linked to",
)
# Timestamps
created_at: datetime = Field(
default_factory=lambda: datetime.now(tz=UTC),
description="Creation timestamp (UTC)",
)
updated_at: datetime = Field(
default_factory=lambda: datetime.now(tz=UTC),
description="Last update timestamp (UTC)",
)
model_config = ConfigDict(
frozen=True,
str_strip_whitespace=True,
use_enum_values=True,
)
# -- Validators ----------------------------------------------------------
@field_validator("name")
@classmethod
def validate_name(cls: type[Resource], v: str | None) -> str | None:
"""Validate namespaced name format if provided.
Names follow the pattern: [[server:]namespace/]name
Built-in resource types may be unnamespaced (e.g., 'git-checkout').
"""
if v is None:
return v
if not v.strip():
raise ValueError("Resource name cannot be empty (use None instead)")
return v
@field_validator("location")
@classmethod
def validate_location(cls: type[Resource], v: str | None) -> str | None:
"""Validate location is non-empty if provided."""
if v is not None and not v.strip():
raise ValueError("Location cannot be empty (use None instead)")
return v
@field_validator("resource_type_name")
@classmethod
def validate_resource_type_name(cls: type[Resource], v: str) -> str:
"""Validate resource type name is non-empty."""
if not v.strip():
raise ValueError("resource_type_name cannot be empty")
return v
# -- Domain Methods ------------------------------------------------------
@property
def is_physical(self) -> bool:
"""Check if this is a physical resource."""
return self.classification == PhysVirt.PHYSICAL
@property
def is_virtual(self) -> bool:
"""Check if this is a virtual resource."""
return self.classification == PhysVirt.VIRTUAL
@property
def can_sandbox(self) -> bool:
"""Check if this resource supports sandbox isolation."""
return self.capabilities.sandboxable
@property
def can_checkpoint(self) -> bool:
"""Check if this resource supports checkpoint/rollback."""
return self.capabilities.checkpointable
@property
def is_read_only(self) -> bool:
"""Check if this resource is read-only."""
return not self.capabilities.writable
def with_updated_at(self) -> Resource:
"""Return a copy with updated_at set to now (UTC)."""
return self.model_copy(update={"updated_at": datetime.now(tz=UTC)})