feat(db): add resource registry tables
CI / lint (push) Waiting to run
CI / typecheck (push) Waiting to run
CI / security (push) Waiting to run
CI / quality (push) Waiting to run
CI / unit_tests (push) Waiting to run
CI / integration_tests (push) Waiting to run
CI / coverage (push) Blocked by required conditions
CI / build (push) Waiting to run
CI / docker (push) Blocked by required conditions
CI / lint (pull_request) Successful in 15s
CI / typecheck (pull_request) Successful in 26s
CI / security (pull_request) Successful in 21s
CI / quality (pull_request) Successful in 15s
CI / integration_tests (pull_request) Successful in 4m16s
CI / build (pull_request) Successful in 16s
CI / unit_tests (pull_request) Successful in 8m37s
CI / coverage (pull_request) Successful in 6m40s
CI / docker (pull_request) Successful in 38s

This commit is contained in:
2026-02-13 23:30:15 +00:00
parent dfb91781aa
commit da8e5f716d
8 changed files with 2150 additions and 76 deletions
@@ -0,0 +1,279 @@
"""Add resource registry tables (resource_types, resources, resource_edges).
This migration creates the three core resource registry tables needed for
the Unified Resource Abstraction Layer:
- ``resource_types``: Schema-level definitions that constrain categories
of resources (CLI args, parent/child rules, sandbox strategy, handler).
- ``resources``: Independently registered entities (git repos, filesystems,
databases, APIs) with ULID primary keys and optional namespaced names.
- ``resource_edges``: DAG edges (parent → child) with type constraints
and link-type annotation.
Revision ID: b1_001_resource_registry
Revises: a5_004_spec_aligned_plans
Create Date: 2026-02-13 12:00:00
"""
from collections.abc import Sequence
import sqlalchemy as sa
from alembic import op
# revision identifiers, used by Alembic.
revision: str = "b1_001_resource_registry"
down_revision: str | Sequence[str] | None = "a5_004_spec_aligned_plans"
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None
def upgrade() -> None:
"""Create resource_types, resources, and resource_edges tables."""
# --- resource_types table ---
op.create_table(
"resource_types",
# PK: namespaced type name (e.g., "builtin/git-checkout")
sa.Column("name", sa.String(255), nullable=False),
# Extracted namespace for filtering (e.g., "builtin")
sa.Column("namespace", sa.String(100), nullable=False),
# Human-readable description
sa.Column("description", sa.Text(), nullable=True),
# Physical or virtual classification
sa.Column("resource_kind", sa.String(20), nullable=False),
# Default sandbox strategy for instances of this type
sa.Column("sandbox_strategy", sa.String(50), nullable=True),
# Whether users can add instances via CLI
sa.Column(
"user_addable",
sa.Boolean(),
nullable=False,
server_default=sa.text("0"),
),
# Handler implementation reference (dotted path or plugin name)
sa.Column("handler_ref", sa.String(500), nullable=True),
# JSON: CLI argument definitions for `resource add <type>`
sa.Column("args_schema_json", sa.Text(), nullable=True),
# JSON: list of allowed parent type names
sa.Column("allowed_parent_types_json", sa.Text(), nullable=True),
# JSON: list of allowed child type definitions
sa.Column("allowed_child_types_json", sa.Text(), nullable=True),
# JSON: auto-discovery configuration
sa.Column("auto_discover_json", sa.Text(), nullable=True),
# JSON: capability flags (readable, writable, sandboxable, checkpointable)
sa.Column("capabilities_json", sa.Text(), nullable=True),
# JSON: equivalence criteria for virtual types
sa.Column("equivalence_json", sa.Text(), nullable=True),
# Source: "builtin" or path to YAML config file
sa.Column("source", sa.String(500), nullable=True),
# Timestamps (ISO-8601 strings)
sa.Column("created_at", sa.String(30), nullable=False),
sa.Column("updated_at", sa.String(30), nullable=False),
# Constraints
sa.PrimaryKeyConstraint("name"),
sa.CheckConstraint(
"resource_kind IN ('physical', 'virtual')",
name="ck_resource_types_kind",
),
)
op.create_index(
"ix_resource_types_namespace",
"resource_types",
["namespace"],
unique=False,
)
op.create_index(
"ix_resource_types_kind",
"resource_types",
["resource_kind"],
unique=False,
)
op.create_index(
"ix_resource_types_user_addable",
"resource_types",
["user_addable"],
unique=False,
)
# --- resources table ---
op.create_table(
"resources",
# PK: ULID (26-char string)
sa.Column("resource_id", sa.String(26), nullable=False),
# Optional namespaced name (NULL for auto-discovered children)
sa.Column("namespaced_name", sa.String(255), nullable=True),
# Extracted namespace for filtering
sa.Column("namespace", sa.String(100), nullable=True),
# Type reference
sa.Column(
"type_name",
sa.String(255),
sa.ForeignKey(
"resource_types.name",
ondelete="RESTRICT",
name="fk_resources_type",
),
nullable=False,
),
# Physical or virtual (derived from type, stored for query efficiency)
sa.Column("resource_kind", sa.String(20), nullable=False),
# Location (physical resources only; NULL for virtual)
sa.Column("location", sa.Text(), nullable=True),
# Human-readable description
sa.Column("description", sa.Text(), nullable=True),
# Read-only flag
sa.Column(
"read_only",
sa.Boolean(),
nullable=False,
server_default=sa.text("0"),
),
# Per-resource sandbox strategy override (NULL = use type default)
sa.Column("sandbox_strategy", sa.String(50), nullable=True),
# Content hash for equivalence tracking (virtual types)
sa.Column("content_hash", sa.String(128), nullable=True),
# JSON: type-specific properties (path, branch, url, etc.)
sa.Column("properties_json", sa.Text(), nullable=True),
# JSON: type-specific metadata
sa.Column("metadata_json", sa.Text(), nullable=True),
# Whether auto-discovered (not user-registered)
sa.Column(
"auto_discovered",
sa.Boolean(),
nullable=False,
server_default=sa.text("0"),
),
# Timestamps (ISO-8601 strings)
sa.Column("created_at", sa.String(30), nullable=False),
sa.Column("updated_at", sa.String(30), nullable=False),
# Constraints
sa.PrimaryKeyConstraint("resource_id"),
sa.CheckConstraint(
"resource_kind IN ('physical', 'virtual')",
name="ck_resources_kind",
),
)
# Unique index on namespaced_name when non-null
op.create_index(
"ix_resources_name",
"resources",
["namespaced_name"],
unique=True,
postgresql_where=sa.text("namespaced_name IS NOT NULL"),
)
# For SQLite: partial unique index using sqlite_where
# Note: SQLAlchemy handles this via the unique=True + nullable column combo;
# SQLite treats NULL values as distinct in unique constraints naturally.
op.create_index(
"ix_resources_namespace",
"resources",
["namespace"],
unique=False,
)
op.create_index(
"ix_resources_type",
"resources",
["type_name"],
unique=False,
)
op.create_index(
"ix_resources_kind",
"resources",
["resource_kind"],
unique=False,
)
op.create_index(
"ix_resources_content_hash",
"resources",
["content_hash"],
unique=False,
)
# --- resource_edges table (DAG: parent -> child) ---
op.create_table(
"resource_edges",
# Parent resource ULID
sa.Column(
"parent_id",
sa.String(26),
sa.ForeignKey(
"resources.resource_id",
ondelete="CASCADE",
name="fk_resource_edges_parent",
),
nullable=False,
),
# Child resource ULID
sa.Column(
"child_id",
sa.String(26),
sa.ForeignKey(
"resources.resource_id",
ondelete="CASCADE",
name="fk_resource_edges_child",
),
nullable=False,
),
# Link type: contains | references | derived_from
sa.Column(
"link_type",
sa.String(30),
nullable=False,
server_default="contains",
),
# Whether auto-discovered (vs manually linked)
sa.Column(
"auto_discovered",
sa.Boolean(),
nullable=False,
server_default=sa.text("0"),
),
# Timestamp
sa.Column("created_at", sa.String(30), nullable=False),
# Constraints
sa.PrimaryKeyConstraint("parent_id", "child_id"),
sa.CheckConstraint(
"parent_id != child_id",
name="ck_resource_edges_no_self_loop",
),
sa.CheckConstraint(
"link_type IN ('contains', 'references', 'derived_from')",
name="ck_resource_edges_link_type",
),
)
op.create_index(
"ix_resource_edges_child",
"resource_edges",
["child_id"],
unique=False,
)
op.create_index(
"ix_resource_edges_link_type",
"resource_edges",
["link_type"],
unique=False,
)
def downgrade() -> None:
"""Drop resource registry tables."""
# Drop in reverse FK order: edges -> resources -> resource_types
op.drop_index("ix_resource_edges_link_type", table_name="resource_edges")
op.drop_index("ix_resource_edges_child", table_name="resource_edges")
op.drop_table("resource_edges")
op.drop_index("ix_resources_content_hash", table_name="resources")
op.drop_index("ix_resources_kind", table_name="resources")
op.drop_index("ix_resources_type", table_name="resources")
op.drop_index("ix_resources_namespace", table_name="resources")
op.drop_index("ix_resources_name", table_name="resources")
op.drop_table("resources")
op.drop_index("ix_resource_types_user_addable", table_name="resource_types")
op.drop_index("ix_resource_types_kind", table_name="resource_types")
op.drop_index("ix_resource_types_namespace", table_name="resource_types")
op.drop_table("resource_types")
@@ -0,0 +1,197 @@
"""ASV benchmarks for resource registry migration and ORM operations.
Measures:
- Schema creation time (Base.metadata.create_all)
- ResourceTypeModel insert throughput
- ResourceModel insert throughput
- ResourceEdgeModel insert throughput
- DAG query performance (children/parents lookup)
"""
from __future__ import annotations
import json
from datetime import UTC, datetime
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from cleveragents.infrastructure.database.models import (
Base,
ResourceEdgeModel,
ResourceModel,
ResourceTypeModel,
)
def _now_iso() -> str:
return datetime.now(tz=UTC).isoformat()
def _make_ulid(n: int) -> str:
return str(n).zfill(26)
class SchemaCreation:
"""Benchmark schema creation (all tables via metadata.create_all)."""
def time_create_all_tables(self) -> None:
engine = create_engine("sqlite:///:memory:")
Base.metadata.create_all(engine)
engine.dispose()
class ResourceTypeInsert:
"""Benchmark inserting resource type records."""
def setup(self) -> None:
self.engine = create_engine("sqlite:///:memory:")
Base.metadata.create_all(self.engine)
self.session_factory = sessionmaker(bind=self.engine)
def teardown(self) -> None:
self.engine.dispose()
def time_insert_single_resource_type(self) -> None:
session = self.session_factory()
rt = ResourceTypeModel()
rt.name = "bench/type-single"
rt.namespace = "bench"
rt.resource_kind = "physical"
rt.user_addable = True
rt.created_at = _now_iso()
rt.updated_at = _now_iso()
session.add(rt)
session.commit()
session.close()
def time_insert_100_resource_types(self) -> None:
session = self.session_factory()
for i in range(100):
rt = ResourceTypeModel()
rt.name = f"bench/type-{i}"
rt.namespace = "bench"
rt.resource_kind = "physical" if i % 2 == 0 else "virtual"
rt.user_addable = i % 3 == 0
rt.args_schema_json = json.dumps([{"flag": f"--arg-{i}"}])
rt.created_at = _now_iso()
rt.updated_at = _now_iso()
session.add(rt)
session.commit()
session.close()
class ResourceInsert:
"""Benchmark inserting resource records."""
def setup(self) -> None:
self.engine = create_engine("sqlite:///:memory:")
Base.metadata.create_all(self.engine)
self.session_factory = sessionmaker(bind=self.engine)
session = self.session_factory()
rt = ResourceTypeModel()
rt.name = "bench/git-checkout"
rt.namespace = "bench"
rt.resource_kind = "physical"
rt.user_addable = True
rt.created_at = _now_iso()
rt.updated_at = _now_iso()
session.add(rt)
session.commit()
session.close()
def teardown(self) -> None:
self.engine.dispose()
def time_insert_100_resources(self) -> None:
session = self.session_factory()
for i in range(100):
r = ResourceModel()
r.resource_id = _make_ulid(i)
r.namespaced_name = f"bench/resource-{i}"
r.namespace = "bench"
r.type_name = "bench/git-checkout"
r.resource_kind = "physical"
r.location = f"/path/to/resource-{i}"
r.properties_json = json.dumps({"index": i})
r.created_at = _now_iso()
r.updated_at = _now_iso()
session.add(r)
session.commit()
session.close()
class ResourceEdgeQuery:
"""Benchmark DAG edge queries (children/parents lookup)."""
def setup(self) -> None:
self.engine = create_engine("sqlite:///:memory:")
Base.metadata.create_all(self.engine)
self.session_factory = sessionmaker(bind=self.engine)
session = self.session_factory()
# Create types
for tname in ("bench/parent-type", "bench/child-type"):
rt = ResourceTypeModel()
rt.name = tname
rt.namespace = "bench"
rt.resource_kind = "physical"
rt.user_addable = True
rt.created_at = _now_iso()
rt.updated_at = _now_iso()
session.add(rt)
# Create 1 parent + 50 children with edges
parent = ResourceModel()
parent.resource_id = _make_ulid(0)
parent.namespaced_name = "bench/parent"
parent.namespace = "bench"
parent.type_name = "bench/parent-type"
parent.resource_kind = "physical"
parent.created_at = _now_iso()
parent.updated_at = _now_iso()
session.add(parent)
self.parent_id = parent.resource_id
for i in range(1, 51):
child = ResourceModel()
child.resource_id = _make_ulid(i)
child.type_name = "bench/child-type"
child.resource_kind = "physical"
child.created_at = _now_iso()
child.updated_at = _now_iso()
session.add(child)
edge = ResourceEdgeModel()
edge.parent_id = parent.resource_id
edge.child_id = child.resource_id
edge.link_type = "contains"
edge.created_at = _now_iso()
session.add(edge)
session.commit()
session.close()
def teardown(self) -> None:
self.engine.dispose()
def time_query_50_children(self) -> None:
session = self.session_factory()
edges = (
session.query(ResourceEdgeModel)
.filter(ResourceEdgeModel.parent_id == self.parent_id)
.all()
)
assert len(edges) == 50
session.close()
def time_query_parent_of_child(self) -> None:
session = self.session_factory()
child_id = _make_ulid(25)
edges = (
session.query(ResourceEdgeModel)
.filter(ResourceEdgeModel.child_id == child_id)
.all()
)
assert len(edges) == 1
session.close()
+124
View File
@@ -272,6 +272,115 @@ Plan-level invariant constraints with scope provenance. Cascades on plan delete.
---
---
## Table: `resource_types`
Schema-level definitions that constrain categories of resources. Built-in types
(e.g. `builtin/git-checkout`) are registered at startup; custom types are loaded
from YAML configuration files.
**Primary Key:** `name`
| Column | Type | Nullable | Default | Description |
|---|---|---|---|---|
| `name` | String(255) | No | -- | PK. Namespaced type name (e.g. `builtin/git-checkout`) |
| `namespace` | String(100) | No | -- | Extracted namespace component |
| `description` | Text | Yes | NULL | Human-readable description |
| `resource_kind` | String(20) | No | -- | `physical` or `virtual` |
| `sandbox_strategy` | String(50) | Yes | NULL | Default sandbox strategy for instances |
| `user_addable` | Boolean | No | `0` | Whether users can add instances via CLI |
| `handler_ref` | String(500) | Yes | NULL | Handler implementation reference |
| `args_schema_json` | Text | Yes | NULL | JSON: CLI argument definitions |
| `allowed_parent_types_json` | Text | Yes | NULL | JSON: allowed parent type names |
| `allowed_child_types_json` | Text | Yes | NULL | JSON: allowed child type definitions |
| `auto_discover_json` | Text | Yes | NULL | JSON: auto-discovery configuration |
| `capabilities_json` | Text | Yes | NULL | JSON: capability flags |
| `equivalence_json` | Text | Yes | NULL | JSON: equivalence criteria (virtual types) |
| `source` | String(500) | Yes | NULL | `builtin` or path to YAML config |
| `created_at` | String(30) | No | -- | ISO-8601 creation timestamp |
| `updated_at` | String(30) | No | -- | ISO-8601 last-update timestamp |
**Check Constraints:**
- `ck_resource_types_kind`: `resource_kind IN ('physical', 'virtual')`
**Indexes:**
- `ix_resource_types_namespace` on `namespace`
- `ix_resource_types_kind` on `resource_kind`
- `ix_resource_types_user_addable` on `user_addable`
---
## Table: `resources`
Independently registered resources representing anything that can be read,
written, or queried. Each resource has a ULID primary key and an optional
namespaced name (auto-discovered children have no name).
**Primary Key:** `resource_id` (ULID, String(26))
| Column | Type | Nullable | Default | Description |
|---|---|---|---|---|
| `resource_id` | String(26) | No | -- | ULID primary key |
| `namespaced_name` | String(255) | Yes | NULL | Optional namespaced name (unique when non-null) |
| `namespace` | String(100) | Yes | NULL | Extracted namespace component |
| `type_name` | String(255) | No | -- | FK to `resource_types.name` (RESTRICT) |
| `resource_kind` | String(20) | No | -- | `physical` or `virtual` (denormalized from type) |
| `location` | Text | Yes | NULL | Physical location (path, URL, etc.) |
| `description` | Text | Yes | NULL | Human-readable description |
| `read_only` | Boolean | No | `0` | Read-only flag |
| `auto_discovered` | Boolean | No | `0` | Whether auto-discovered (not user-registered) |
| `sandbox_strategy` | String(50) | Yes | NULL | Per-resource sandbox strategy override |
| `content_hash` | String(128) | Yes | NULL | Content hash for equivalence tracking |
| `properties_json` | Text | Yes | NULL | JSON: type-specific properties |
| `metadata_json` | Text | Yes | NULL | JSON: type-specific metadata |
| `created_at` | String(30) | No | -- | ISO-8601 creation timestamp |
| `updated_at` | String(30) | No | -- | ISO-8601 last-update timestamp |
**Foreign Keys:**
- `fk_resources_type`: `type_name` -> `resource_types.name` ON DELETE RESTRICT
**Check Constraints:**
- `ck_resources_kind`: `resource_kind IN ('physical', 'virtual')`
**Indexes:**
- `ix_resources_name` on `namespaced_name` (unique, partial - non-null only)
- `ix_resources_namespace` on `namespace`
- `ix_resources_type` on `type_name`
- `ix_resources_kind` on `resource_kind`
- `ix_resources_content_hash` on `content_hash`
---
## Table: `resource_edges`
DAG edges representing parent-child relationships between resources. A resource
can have multiple parents and multiple children, subject to type constraints.
**Primary Key:** `(parent_id, child_id)` (composite)
| Column | Type | Nullable | Default | Description |
|---|---|---|---|---|
| `parent_id` | String(26) | No | -- | FK to `resources.resource_id` (CASCADE) |
| `child_id` | String(26) | No | -- | FK to `resources.resource_id` (CASCADE) |
| `link_type` | String(30) | No | `'contains'` | `contains`, `references`, or `derived_from` |
| `auto_discovered` | Boolean | No | `0` | Whether auto-discovered (vs manually linked) |
| `created_at` | String(30) | No | -- | ISO-8601 timestamp |
**Foreign Keys:**
- `fk_resource_edges_parent`: `parent_id` -> `resources.resource_id` ON DELETE CASCADE
- `fk_resource_edges_child`: `child_id` -> `resources.resource_id` ON DELETE CASCADE
**Check Constraints:**
- `ck_resource_edges_no_self_loop`: `parent_id != child_id`
- `ck_resource_edges_link_type`: `link_type IN ('contains', 'references', 'derived_from')`
**Indexes:**
- `ix_resource_edges_child` on `child_id`
- `ix_resource_edges_link_type` on `link_type`
---
## Entity Relationship Diagram
```
@@ -286,11 +395,26 @@ v3_plans (PK: plan_id)
|-- 1:N --> plan_projects (FK: plan_id, CASCADE)
|-- 1:N --> plan_arguments (FK: plan_id, CASCADE)
|-- 1:N --> plan_invariants (FK: plan_id, CASCADE)
resource_types (PK: name)
|-- 1:N --> resources (FK: type_name, RESTRICT)
resources (PK: resource_id)
|-- N:M --> resource_edges (FK: parent_id/child_id, CASCADE)
```
## Migration Chain
| Revision | Description |
|---|---|
| `a5_003_spec_aligned_actions` | Drops `lifecycle_plans` and `actions_v3`; creates `actions`, `action_invariants`, `action_arguments` |
| `a5_004_spec_aligned_plans` | Creates `v3_plans`, `plan_projects`, `plan_arguments`, `plan_invariants` |
| `b1_001_resource_registry` | Creates `resource_types`, `resources`, `resource_edges` |
## Source Location
- Migration (actions): `alembic/versions/a5_003_spec_aligned_actions.py`
- Migration (plans): `alembic/versions/a5_004_spec_aligned_plans.py`
- Migration (resources): `alembic/versions/b1_001_resource_registry_tables.py`
- ORM models: `src/cleveragents/infrastructure/database/models.py`
- Repositories: `src/cleveragents/infrastructure/database/repositories.py`
+236
View File
@@ -0,0 +1,236 @@
Feature: Resource registry database tables
As a developer working on the resource registry infrastructure
I need database tables for resource types, resources, and resource edges
So that the resource DAG can be persisted and queried
Background:
Given a resource registry in-memory database is initialized
# ---------------------------------------------------------------------------
# Table existence
# ---------------------------------------------------------------------------
Scenario: Resource registry tables exist after schema creation
Then the resource registry database should contain table "resource_types"
And the resource registry database should contain table "resources"
And the resource registry database should contain table "resource_edges"
# ---------------------------------------------------------------------------
# ResourceTypeModel CRUD
# ---------------------------------------------------------------------------
Scenario: Persist and retrieve a physical resource type
When I create a resource type "builtin/git-checkout" with kind "physical" and user_addable true
Then I can retrieve resource type "builtin/git-checkout"
And the retrieved resource type namespace is "builtin"
And the retrieved resource type kind is "physical"
And the retrieved resource type user_addable is true
Scenario: Persist and retrieve a virtual resource type
When I create a resource type "builtin/file" with kind "virtual" and user_addable false
Then I can retrieve resource type "builtin/file"
And the retrieved resource type kind is "virtual"
And the retrieved resource type user_addable is false
Scenario: Resource type stores JSON fields correctly
When I create a resource type "builtin/git-checkout" with JSON fields
Then the resource type args_schema_json round-trips correctly
And the resource type allowed_parent_types_json round-trips correctly
And the resource type allowed_child_types_json round-trips correctly
And the resource type auto_discover_json round-trips correctly
And the resource type capabilities_json round-trips correctly
Scenario: Resource type name uniqueness is enforced
When I create a resource type "builtin/git-checkout" with kind "physical" and user_addable true
Then creating a duplicate resource type "builtin/git-checkout" raises an integrity error
Scenario: Resource type kind check constraint rejects invalid values
Then creating a resource type with kind "invalid_kind" raises an integrity error
Scenario: List resource types by namespace
When I create a resource type "builtin/git-checkout" with kind "physical" and user_addable true
And I create a resource type "builtin/fs-mount" with kind "physical" and user_addable true
And I create a resource type "custom/database" with kind "physical" and user_addable true
Then listing resource types by namespace "builtin" returns 2 results
And listing resource types by namespace "custom" returns 1 result
Scenario: List resource types by kind
When I create a resource type "builtin/git-checkout" with kind "physical" and user_addable true
And I create a resource type "builtin/file" with kind "virtual" and user_addable false
Then listing resource types by kind "physical" returns 1 result
And listing resource types by kind "virtual" returns 1 result
# ---------------------------------------------------------------------------
# ResourceModel CRUD
# ---------------------------------------------------------------------------
Scenario: Persist and retrieve a named resource
Given a resource type "builtin/git-checkout" exists
When I create a resource "local/my-repo" of type "builtin/git-checkout" with location "/home/user/repo"
Then I can retrieve the resource by id
And the retrieved resource namespaced_name is "local/my-repo"
And the retrieved resource namespace is "local"
And the retrieved resource type_name is "builtin/git-checkout"
And the retrieved resource kind is "physical"
And the retrieved resource location is "/home/user/repo"
Scenario: Persist a resource without a name (auto-discovered child)
Given a resource type "builtin/fs-file" exists as virtual
When I create an unnamed resource of type "builtin/fs-file"
Then I can retrieve the unnamed resource by id
And the unnamed resource namespaced_name is null
Scenario: Resource stores JSON properties correctly
Given a resource type "builtin/git-checkout" exists
When I create a resource with JSON properties
Then the resource properties_json round-trips correctly
And the resource metadata_json round-trips correctly
Scenario: Resource kind check constraint rejects invalid values
Given a resource type "builtin/git-checkout" exists
Then creating a resource with kind "invalid_kind" raises an integrity error
Scenario: Resource type FK prevents orphaned resources
Then creating a resource with non-existent type raises an integrity error
Scenario: Named resource uniqueness is enforced
Given a resource type "builtin/git-checkout" exists
When I create a resource "local/my-repo" of type "builtin/git-checkout" with location "/path/a"
Then creating a duplicate named resource "local/my-repo" raises an integrity error
Scenario: Resource RESTRICT prevents type deletion when resources exist
Given a resource type "builtin/git-checkout" exists
When I create a resource "local/my-repo" of type "builtin/git-checkout" with location "/path"
Then deleting resource type "builtin/git-checkout" raises an integrity error
Scenario: Query resources by type
Given a resource type "builtin/git-checkout" exists
And a resource type "builtin/fs-mount" exists as physical
When I create a resource "local/repo1" of type "builtin/git-checkout" with location "/repo1"
And I create a resource "local/mount1" of type "builtin/fs-mount" with location "/mnt"
Then querying resources by type "builtin/git-checkout" returns 1 result
And querying resources by type "builtin/fs-mount" returns 1 result
Scenario: Query resources by namespace
Given a resource type "builtin/git-checkout" exists
When I create a resource "local/repo1" of type "builtin/git-checkout" with location "/repo1"
And I create a resource "team/repo2" of type "builtin/git-checkout" with location "/repo2"
Then querying resources by namespace "local" returns 1 result
And querying resources by namespace "team" returns 1 result
# ---------------------------------------------------------------------------
# ResourceEdgeModel (DAG edges)
# ---------------------------------------------------------------------------
Scenario: Create and query parent-child edges
Given a resource type "builtin/git-checkout" exists
And a resource type "builtin/fs-directory" exists as physical
And a parent resource "local/repo" of type "builtin/git-checkout"
And a child resource of type "builtin/fs-directory"
When I create an edge from parent to child with link_type "contains"
Then the edge exists in the database
And the edge link_type is "contains"
And querying children of the parent returns 1 result
And querying parents of the child returns 1 result
Scenario: Edge composite PK prevents duplicate edges
Given a resource type "builtin/git-checkout" exists
And a resource type "builtin/fs-directory" exists as physical
And a parent resource "local/repo" of type "builtin/git-checkout"
And a child resource of type "builtin/fs-directory"
When I create an edge from parent to child with link_type "contains"
Then creating a duplicate edge raises an integrity error
Scenario: Self-loop check constraint prevents self-referencing edges
Given a resource type "builtin/git-checkout" exists
And a parent resource "local/repo" of type "builtin/git-checkout"
Then creating a self-loop edge raises an integrity error
Scenario: Edge link_type check constraint rejects invalid values
Given a resource type "builtin/git-checkout" exists
And a resource type "builtin/fs-directory" exists as physical
And a parent resource "local/repo" of type "builtin/git-checkout"
And a child resource of type "builtin/fs-directory"
Then creating an edge with invalid link_type raises an integrity error
Scenario: CASCADE deletes edges when parent resource is deleted
Given a resource type "builtin/git-checkout" exists
And a resource type "builtin/fs-directory" exists as physical
And a parent resource "local/repo" of type "builtin/git-checkout"
And a child resource of type "builtin/fs-directory"
And I create an edge from parent to child with link_type "contains"
When I delete the parent resource
Then the edge no longer exists in the database
Scenario: CASCADE deletes edges when child resource is deleted
Given a resource type "builtin/git-checkout" exists
And a resource type "builtin/fs-directory" exists as physical
And a parent resource "local/repo" of type "builtin/git-checkout"
And a child resource of type "builtin/fs-directory"
And I create an edge from parent to child with link_type "contains"
When I delete the child resource
Then the edge no longer exists in the database
Scenario: Multiple parents for a single child resource
Given a resource type "builtin/git-checkout" exists
And a resource type "builtin/fs-directory" exists as physical
And a parent resource "local/repo-a" of type "builtin/git-checkout"
And a second parent resource "local/repo-b" of type "builtin/git-checkout"
And a child resource of type "builtin/fs-directory"
When I create an edge from parent to child with link_type "contains"
And I create an edge from second parent to child with link_type "references"
Then querying parents of the child returns 2 results
Scenario: Multiple children for a single parent resource
Given a resource type "builtin/git-checkout" exists
And a resource type "builtin/fs-directory" exists as physical
And a parent resource "local/repo" of type "builtin/git-checkout"
And a first child resource of type "builtin/fs-directory"
And a second child resource of type "builtin/fs-directory"
When I create edges from parent to both children
Then querying children of the parent returns 2 results
# ---------------------------------------------------------------------------
# ORM relationship navigation
# ---------------------------------------------------------------------------
Scenario: Navigate from resource to resource type via relationship
Given a resource type "builtin/git-checkout" exists
When I create a resource "local/repo" of type "builtin/git-checkout" with location "/repo"
Then navigating from resource to resource_type returns "builtin/git-checkout"
Scenario: Navigate from resource type to resources via relationship
Given a resource type "builtin/git-checkout" exists
When I create a resource "local/repo1" of type "builtin/git-checkout" with location "/r1"
And I create a resource "local/repo2" of type "builtin/git-checkout" with location "/r2"
Then navigating from resource_type to resources returns 2 items
Scenario: Navigate from resource to child edges via relationship
Given a resource type "builtin/git-checkout" exists
And a resource type "builtin/fs-directory" exists as physical
And a parent resource "local/repo" of type "builtin/git-checkout"
And a child resource of type "builtin/fs-directory"
And I create an edge from parent to child with link_type "contains"
Then navigating from parent to child_edges returns 1 edge
And navigating from child to parent_edges returns 1 edge
# ---------------------------------------------------------------------------
# Timestamps and metadata
# ---------------------------------------------------------------------------
Scenario: Resource type timestamps are stored correctly
When I create a resource type "builtin/git-checkout" with kind "physical" and user_addable true
Then the resource type has valid created_at and updated_at timestamps
Scenario: Resource timestamps are stored correctly
Given a resource type "builtin/git-checkout" exists
When I create a resource "local/repo" of type "builtin/git-checkout" with location "/repo"
Then the resource has valid created_at and updated_at timestamps
Scenario: Edge timestamp is stored correctly
Given a resource type "builtin/git-checkout" exists
And a resource type "builtin/fs-directory" exists as physical
And a parent resource "local/repo" of type "builtin/git-checkout"
And a child resource of type "builtin/fs-directory"
When I create an edge from parent to child with link_type "contains"
Then the edge has a valid created_at timestamp
@@ -0,0 +1,943 @@
"""Step definitions for resource registry database table tests.
Tests the ResourceTypeModel, ResourceModel, and ResourceEdgeModel ORM models
and their corresponding database tables (resource_types, resources,
resource_edges). Uses an in-memory SQLite database with schema created via
``Base.metadata.create_all``.
"""
from __future__ import annotations
import json
from datetime import UTC, datetime
from typing import Any
from behave import given, then, when # type: ignore[import-untyped]
from sqlalchemy import create_engine, event, inspect, text
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import Session, sessionmaker
from cleveragents.infrastructure.database.models import (
Base,
ResourceEdgeModel,
ResourceModel,
ResourceTypeModel,
)
def _now_iso() -> str:
"""Return current UTC time as ISO-8601 string."""
return datetime.now(tz=UTC).isoformat()
def _make_ulid(suffix: str = "0") -> str:
"""Create a deterministic 26-char fake ULID for testing."""
return ("0" * (26 - len(suffix))) + suffix
# ---------------------------------------------------------------------------
# Background
# ---------------------------------------------------------------------------
def _set_sqlite_pragma(dbapi_conn: Any, _connection_record: Any) -> None:
"""Enable FK enforcement on every new SQLite connection."""
cursor = dbapi_conn.cursor()
cursor.execute("PRAGMA foreign_keys = ON")
cursor.close()
@given("a resource registry in-memory database is initialized")
def step_resource_registry_init_db(context: Any) -> None:
"""Create an in-memory SQLite database with all tables."""
engine = create_engine("sqlite:///:memory:")
# Enable FK enforcement on every connection via event listener
event.listen(engine, "connect", _set_sqlite_pragma)
# Re-establish FK pragma on the initial connection used for metadata
with engine.connect() as conn:
conn.execute(text("PRAGMA foreign_keys = ON"))
conn.commit()
Base.metadata.create_all(engine)
context.rr_engine = engine
context.rr_session_factory = sessionmaker(bind=engine)
context.rr_session = context.rr_session_factory()
# Storage for cross-step references
context.rr_resource_type = None
context.rr_resource = None
context.rr_parent_resource = None
context.rr_parent_resource_2 = None
context.rr_child_resource = None
context.rr_child_resource_2 = None
context.rr_edge = None
context.rr_error = None
# ---------------------------------------------------------------------------
# Table existence
# ---------------------------------------------------------------------------
@then('the resource registry database should contain table "{table_name}"')
def step_resource_registry_table_exists(context: Any, table_name: str) -> None:
inspector = inspect(context.rr_engine)
tables = inspector.get_table_names()
assert table_name in tables, f"Table {table_name} not found. Tables: {tables}"
# ---------------------------------------------------------------------------
# ResourceTypeModel CRUD
# ---------------------------------------------------------------------------
@when('I create a resource type "{name}" with kind "{kind}" and user_addable {addable}')
def step_resource_registry_create_type(
context: Any, name: str, kind: str, addable: str
) -> None:
session: Session = context.rr_session
namespace = name.split("/")[0]
rt = ResourceTypeModel()
rt.name = name
rt.namespace = namespace
rt.resource_kind = kind
rt.user_addable = addable.lower() == "true"
rt.created_at = _now_iso()
rt.updated_at = _now_iso()
session.add(rt)
session.commit()
context.rr_resource_type = rt
@then('I can retrieve resource type "{name}"')
def step_resource_registry_get_type(context: Any, name: str) -> None:
session: Session = context.rr_session
rt = session.get(ResourceTypeModel, name)
assert rt is not None, f"ResourceType {name} not found"
context.rr_resource_type = rt
@then('the retrieved resource type namespace is "{expected}"')
def step_resource_registry_type_namespace(context: Any, expected: str) -> None:
assert context.rr_resource_type.namespace == expected
@then('the retrieved resource type kind is "{expected}"')
def step_resource_registry_type_kind(context: Any, expected: str) -> None:
assert context.rr_resource_type.resource_kind == expected
@then("the retrieved resource type user_addable is {expected}")
def step_resource_registry_type_user_addable(context: Any, expected: str) -> None:
expected_bool = expected.lower() == "true"
assert context.rr_resource_type.user_addable == expected_bool
@when('I create a resource type "{name}" with JSON fields')
def step_resource_registry_create_type_json(context: Any, name: str) -> None:
session: Session = context.rr_session
namespace = name.split("/")[0]
rt = ResourceTypeModel()
rt.name = name
rt.namespace = namespace
rt.resource_kind = "physical"
rt.user_addable = True
rt.args_schema_json = json.dumps(
[{"flag": "--path", "help": "Path to repo", "required": True}]
)
rt.allowed_parent_types_json = json.dumps([])
rt.allowed_child_types_json = json.dumps(
[{"type": "builtin/fs-directory"}, {"type": "builtin/git"}]
)
rt.auto_discover_json = json.dumps({"enabled": True, "max_depth": 10})
rt.capabilities_json = json.dumps(
{
"readable": True,
"writable": True,
"sandboxable": True,
"checkpointable": True,
}
)
rt.created_at = _now_iso()
rt.updated_at = _now_iso()
session.add(rt)
session.commit()
context.rr_resource_type = rt
@then("the resource type args_schema_json round-trips correctly")
def step_resource_registry_type_args_json(context: Any) -> None:
data = json.loads(context.rr_resource_type.args_schema_json)
assert isinstance(data, list)
assert data[0]["flag"] == "--path"
@then("the resource type allowed_parent_types_json round-trips correctly")
def step_resource_registry_type_parents_json(context: Any) -> None:
data = json.loads(context.rr_resource_type.allowed_parent_types_json)
assert isinstance(data, list)
@then("the resource type allowed_child_types_json round-trips correctly")
def step_resource_registry_type_children_json(context: Any) -> None:
data = json.loads(context.rr_resource_type.allowed_child_types_json)
assert isinstance(data, list)
assert len(data) == 2
@then("the resource type auto_discover_json round-trips correctly")
def step_resource_registry_type_autodiscover_json(context: Any) -> None:
data = json.loads(context.rr_resource_type.auto_discover_json)
assert data["enabled"] is True
assert data["max_depth"] == 10
@then("the resource type capabilities_json round-trips correctly")
def step_resource_registry_type_capabilities_json(context: Any) -> None:
data = json.loads(context.rr_resource_type.capabilities_json)
assert data["readable"] is True
assert data["sandboxable"] is True
@then('creating a duplicate resource type "{name}" raises an integrity error')
def step_resource_registry_duplicate_type(context: Any, name: str) -> None:
session: Session = context.rr_session
namespace = name.split("/")[0]
rt = ResourceTypeModel()
rt.name = name
rt.namespace = namespace
rt.resource_kind = "physical"
rt.user_addable = False
rt.created_at = _now_iso()
rt.updated_at = _now_iso()
try:
session.add(rt)
session.commit()
msg = "Expected IntegrityError for duplicate resource type"
raise AssertionError(msg)
except IntegrityError:
session.rollback()
@then('creating a resource type with kind "{kind}" raises an integrity error')
def step_resource_registry_invalid_kind(context: Any, kind: str) -> None:
session: Session = context.rr_session
try:
session.execute(
text(
"INSERT INTO resource_types (name, namespace, resource_kind, "
"user_addable, created_at, updated_at) "
"VALUES (:name, :ns, :kind, 0, :ts, :ts)"
),
{
"name": "test/invalid-kind",
"ns": "test",
"kind": kind,
"ts": _now_iso(),
},
)
session.commit()
msg = "Expected IntegrityError for invalid resource_kind"
raise AssertionError(msg)
except IntegrityError:
session.rollback()
@then('listing resource types by namespace "{ns}" returns {count:d} results')
def step_resource_registry_list_types_ns_plural(
context: Any, ns: str, count: int
) -> None:
session: Session = context.rr_session
results = (
session.query(ResourceTypeModel).filter(ResourceTypeModel.namespace == ns).all()
)
assert len(results) == count, f"Expected {count}, got {len(results)}"
@then('listing resource types by namespace "{ns}" returns {count:d} result')
def step_resource_registry_list_types_ns_singular(
context: Any, ns: str, count: int
) -> None:
session: Session = context.rr_session
results = (
session.query(ResourceTypeModel).filter(ResourceTypeModel.namespace == ns).all()
)
assert len(results) == count, f"Expected {count}, got {len(results)}"
@then('listing resource types by kind "{kind}" returns {count:d} results')
def step_resource_registry_list_types_kind_plural(
context: Any, kind: str, count: int
) -> None:
session: Session = context.rr_session
results = (
session.query(ResourceTypeModel)
.filter(ResourceTypeModel.resource_kind == kind)
.all()
)
assert len(results) == count, f"Expected {count}, got {len(results)}"
@then('listing resource types by kind "{kind}" returns {count:d} result')
def step_resource_registry_list_types_kind_singular(
context: Any, kind: str, count: int
) -> None:
session: Session = context.rr_session
results = (
session.query(ResourceTypeModel)
.filter(ResourceTypeModel.resource_kind == kind)
.all()
)
assert len(results) == count, f"Expected {count}, got {len(results)}"
@then('listing resource types by kind "{kind}" returns {count:d} result')
def step_resource_registry_list_types_kind(context: Any, kind: str, count: int) -> None:
session: Session = context.rr_session
results = (
session.query(ResourceTypeModel)
.filter(ResourceTypeModel.resource_kind == kind)
.all()
)
assert len(results) == count, f"Expected {count}, got {len(results)}"
# ---------------------------------------------------------------------------
# ResourceModel CRUD - prerequisites
# ---------------------------------------------------------------------------
@given('a resource type "{name}" exists')
def step_resource_registry_ensure_type_physical(context: Any, name: str) -> None:
session: Session = context.rr_session
existing = session.get(ResourceTypeModel, name)
if existing is None:
namespace = name.split("/")[0]
rt = ResourceTypeModel()
rt.name = name
rt.namespace = namespace
rt.resource_kind = "physical"
rt.user_addable = True
rt.created_at = _now_iso()
rt.updated_at = _now_iso()
session.add(rt)
session.commit()
@given('a resource type "{name}" exists as virtual')
def step_resource_registry_ensure_type_virtual(context: Any, name: str) -> None:
session: Session = context.rr_session
existing = session.get(ResourceTypeModel, name)
if existing is None:
namespace = name.split("/")[0]
rt = ResourceTypeModel()
rt.name = name
rt.namespace = namespace
rt.resource_kind = "virtual"
rt.user_addable = False
rt.created_at = _now_iso()
rt.updated_at = _now_iso()
session.add(rt)
session.commit()
@given('a resource type "{name}" exists as physical')
def step_resource_registry_ensure_type_physical_explicit(
context: Any, name: str
) -> None:
session: Session = context.rr_session
existing = session.get(ResourceTypeModel, name)
if existing is None:
namespace = name.split("/")[0]
rt = ResourceTypeModel()
rt.name = name
rt.namespace = namespace
rt.resource_kind = "physical"
rt.user_addable = True
rt.created_at = _now_iso()
rt.updated_at = _now_iso()
session.add(rt)
session.commit()
# ---------------------------------------------------------------------------
# ResourceModel CRUD
# ---------------------------------------------------------------------------
@when('I create a resource "{rname}" of type "{tname}" with location "{loc}"')
def step_resource_registry_create_resource(
context: Any, rname: str, tname: str, loc: str
) -> None:
session: Session = context.rr_session
namespace = rname.split("/")[0]
rid = _make_ulid(rname.replace("/", "_"))
# Look up the type to get the kind
rt = session.get(ResourceTypeModel, tname)
kind = rt.resource_kind if rt else "physical"
r = ResourceModel()
r.resource_id = rid
r.namespaced_name = rname
r.namespace = namespace
r.type_name = tname
r.resource_kind = kind
r.location = loc
r.created_at = _now_iso()
r.updated_at = _now_iso()
session.add(r)
session.commit()
context.rr_resource = r
@then("I can retrieve the resource by id")
def step_resource_registry_get_resource(context: Any) -> None:
session: Session = context.rr_session
rid = context.rr_resource.resource_id
r = session.get(ResourceModel, rid)
assert r is not None, f"Resource {rid} not found"
context.rr_resource = r
@then('the retrieved resource namespaced_name is "{expected}"')
def step_resource_registry_resource_name(context: Any, expected: str) -> None:
assert context.rr_resource.namespaced_name == expected
@then('the retrieved resource namespace is "{expected}"')
def step_resource_registry_resource_namespace(context: Any, expected: str) -> None:
assert context.rr_resource.namespace == expected
@then('the retrieved resource type_name is "{expected}"')
def step_resource_registry_resource_type(context: Any, expected: str) -> None:
assert context.rr_resource.type_name == expected
@then('the retrieved resource kind is "{expected}"')
def step_resource_registry_resource_kind(context: Any, expected: str) -> None:
assert context.rr_resource.resource_kind == expected
@then('the retrieved resource location is "{expected}"')
def step_resource_registry_resource_location(context: Any, expected: str) -> None:
assert context.rr_resource.location == expected
@when('I create an unnamed resource of type "{tname}"')
def step_resource_registry_create_unnamed(context: Any, tname: str) -> None:
session: Session = context.rr_session
rid = _make_ulid("unnamed1")
rt = session.get(ResourceTypeModel, tname)
kind = rt.resource_kind if rt else "virtual"
r = ResourceModel()
r.resource_id = rid
r.namespaced_name = None
r.namespace = None
r.type_name = tname
r.resource_kind = kind
r.created_at = _now_iso()
r.updated_at = _now_iso()
session.add(r)
session.commit()
context.rr_resource = r
@then("I can retrieve the unnamed resource by id")
def step_resource_registry_get_unnamed(context: Any) -> None:
session: Session = context.rr_session
rid = context.rr_resource.resource_id
r = session.get(ResourceModel, rid)
assert r is not None, f"Resource {rid} not found"
context.rr_resource = r
@then("the unnamed resource namespaced_name is null")
def step_resource_registry_unnamed_null(context: Any) -> None:
assert context.rr_resource.namespaced_name is None
@when("I create a resource with JSON properties")
def step_resource_registry_create_resource_json(context: Any) -> None:
session: Session = context.rr_session
rid = _make_ulid("jsonprops")
r = ResourceModel()
r.resource_id = rid
r.namespaced_name = "local/json-test"
r.namespace = "local"
r.type_name = "builtin/git-checkout"
r.resource_kind = "physical"
r.location = "/path/to/repo"
r.properties_json = json.dumps({"branch": "main", "remote": "origin"})
r.metadata_json = json.dumps({"tags": ["important"], "priority": 1})
r.created_at = _now_iso()
r.updated_at = _now_iso()
session.add(r)
session.commit()
context.rr_resource = r
@then("the resource properties_json round-trips correctly")
def step_resource_registry_resource_props(context: Any) -> None:
data = json.loads(context.rr_resource.properties_json)
assert data["branch"] == "main"
assert data["remote"] == "origin"
@then("the resource metadata_json round-trips correctly")
def step_resource_registry_resource_meta(context: Any) -> None:
data = json.loads(context.rr_resource.metadata_json)
assert data["tags"] == ["important"]
assert data["priority"] == 1
@then('creating a resource with kind "{kind}" raises an integrity error')
def step_resource_registry_invalid_resource_kind(context: Any, kind: str) -> None:
session: Session = context.rr_session
try:
session.execute(
text(
"INSERT INTO resources (resource_id, type_name, resource_kind, "
"read_only, auto_discovered, created_at, updated_at) "
"VALUES (:rid, :tname, :kind, 0, 0, :ts, :ts)"
),
{
"rid": _make_ulid("badkind"),
"tname": "builtin/git-checkout",
"kind": kind,
"ts": _now_iso(),
},
)
session.commit()
msg = "Expected IntegrityError for invalid resource_kind"
raise AssertionError(msg)
except IntegrityError:
session.rollback()
@then("creating a resource with non-existent type raises an integrity error")
def step_resource_registry_missing_type(context: Any) -> None:
session: Session = context.rr_session
r = ResourceModel()
r.resource_id = _make_ulid("orphan")
r.type_name = "nonexistent/type"
r.resource_kind = "physical"
r.created_at = _now_iso()
r.updated_at = _now_iso()
try:
session.add(r)
session.commit()
msg = "Expected IntegrityError for missing FK"
raise AssertionError(msg)
except IntegrityError:
session.rollback()
@then('creating a duplicate named resource "{name}" raises an integrity error')
def step_resource_registry_duplicate_resource(context: Any, name: str) -> None:
session: Session = context.rr_session
r = ResourceModel()
r.resource_id = _make_ulid("dup")
r.namespaced_name = name
r.namespace = name.split("/")[0]
r.type_name = "builtin/git-checkout"
r.resource_kind = "physical"
r.created_at = _now_iso()
r.updated_at = _now_iso()
try:
session.add(r)
session.commit()
msg = "Expected IntegrityError for duplicate named resource"
raise AssertionError(msg)
except IntegrityError:
session.rollback()
@then('deleting resource type "{name}" raises an integrity error')
def step_resource_registry_restrict_type_delete(context: Any, name: str) -> None:
session: Session = context.rr_session
rt = session.get(ResourceTypeModel, name)
assert rt is not None
try:
session.delete(rt)
session.flush()
msg = "Expected IntegrityError for RESTRICT delete"
raise AssertionError(msg)
except IntegrityError:
session.rollback()
@then('querying resources by type "{tname}" returns {count:d} result')
def step_resource_registry_query_by_type(context: Any, tname: str, count: int) -> None:
session: Session = context.rr_session
results = (
session.query(ResourceModel).filter(ResourceModel.type_name == tname).all()
)
assert len(results) == count, f"Expected {count}, got {len(results)}"
@then('querying resources by namespace "{ns}" returns {count:d} result')
def step_resource_registry_query_by_ns(context: Any, ns: str, count: int) -> None:
session: Session = context.rr_session
results = session.query(ResourceModel).filter(ResourceModel.namespace == ns).all()
assert len(results) == count, f"Expected {count}, got {len(results)}"
# ---------------------------------------------------------------------------
# ResourceEdgeModel - prerequisites
# ---------------------------------------------------------------------------
@given('a parent resource "{rname}" of type "{tname}"')
def step_resource_registry_create_parent(context: Any, rname: str, tname: str) -> None:
session: Session = context.rr_session
rid = _make_ulid("parent_" + rname.replace("/", "_"))
rt = session.get(ResourceTypeModel, tname)
kind = rt.resource_kind if rt else "physical"
r = ResourceModel()
r.resource_id = rid
r.namespaced_name = rname
r.namespace = rname.split("/")[0]
r.type_name = tname
r.resource_kind = kind
r.created_at = _now_iso()
r.updated_at = _now_iso()
session.add(r)
session.commit()
context.rr_parent_resource = r
@given('a second parent resource "{rname}" of type "{tname}"')
def step_resource_registry_create_second_parent(
context: Any, rname: str, tname: str
) -> None:
session: Session = context.rr_session
rid = _make_ulid("parent2_" + rname.replace("/", "_"))
rt = session.get(ResourceTypeModel, tname)
kind = rt.resource_kind if rt else "physical"
r = ResourceModel()
r.resource_id = rid
r.namespaced_name = rname
r.namespace = rname.split("/")[0]
r.type_name = tname
r.resource_kind = kind
r.created_at = _now_iso()
r.updated_at = _now_iso()
session.add(r)
session.commit()
context.rr_parent_resource_2 = r
@given('a child resource of type "{tname}"')
def step_resource_registry_create_child(context: Any, tname: str) -> None:
session: Session = context.rr_session
rid = _make_ulid("child1")
rt = session.get(ResourceTypeModel, tname)
kind = rt.resource_kind if rt else "physical"
r = ResourceModel()
r.resource_id = rid
r.type_name = tname
r.resource_kind = kind
r.created_at = _now_iso()
r.updated_at = _now_iso()
session.add(r)
session.commit()
context.rr_child_resource = r
@given('a first child resource of type "{tname}"')
def step_resource_registry_create_first_child(context: Any, tname: str) -> None:
session: Session = context.rr_session
rid = _make_ulid("child_a")
rt = session.get(ResourceTypeModel, tname)
kind = rt.resource_kind if rt else "physical"
r = ResourceModel()
r.resource_id = rid
r.type_name = tname
r.resource_kind = kind
r.created_at = _now_iso()
r.updated_at = _now_iso()
session.add(r)
session.commit()
context.rr_child_resource = r
@given('a second child resource of type "{tname}"')
def step_resource_registry_create_second_child(context: Any, tname: str) -> None:
session: Session = context.rr_session
rid = _make_ulid("child_b")
rt = session.get(ResourceTypeModel, tname)
kind = rt.resource_kind if rt else "physical"
r = ResourceModel()
r.resource_id = rid
r.type_name = tname
r.resource_kind = kind
r.created_at = _now_iso()
r.updated_at = _now_iso()
session.add(r)
session.commit()
context.rr_child_resource_2 = r
# ---------------------------------------------------------------------------
# ResourceEdgeModel CRUD
# ---------------------------------------------------------------------------
@given('I create an edge from parent to child with link_type "{lt}"')
@when('I create an edge from parent to child with link_type "{lt}"')
def step_resource_registry_create_edge(context: Any, lt: str) -> None:
session: Session = context.rr_session
edge = ResourceEdgeModel()
edge.parent_id = context.rr_parent_resource.resource_id
edge.child_id = context.rr_child_resource.resource_id
edge.link_type = lt
edge.created_at = _now_iso()
session.add(edge)
session.commit()
context.rr_edge = edge
@when('I create an edge from second parent to child with link_type "{lt}"')
def step_resource_registry_create_edge_second_parent(context: Any, lt: str) -> None:
session: Session = context.rr_session
edge = ResourceEdgeModel()
edge.parent_id = context.rr_parent_resource_2.resource_id
edge.child_id = context.rr_child_resource.resource_id
edge.link_type = lt
edge.created_at = _now_iso()
session.add(edge)
session.commit()
@when("I create edges from parent to both children")
def step_resource_registry_create_edges_multi(context: Any) -> None:
session: Session = context.rr_session
e1 = ResourceEdgeModel()
e1.parent_id = context.rr_parent_resource.resource_id
e1.child_id = context.rr_child_resource.resource_id
e1.link_type = "contains"
e1.created_at = _now_iso()
session.add(e1)
e2 = ResourceEdgeModel()
e2.parent_id = context.rr_parent_resource.resource_id
e2.child_id = context.rr_child_resource_2.resource_id
e2.link_type = "contains"
e2.created_at = _now_iso()
session.add(e2)
session.commit()
@then("the edge exists in the database")
def step_resource_registry_edge_exists(context: Any) -> None:
session: Session = context.rr_session
edge = (
session.query(ResourceEdgeModel)
.filter(
ResourceEdgeModel.parent_id == context.rr_parent_resource.resource_id,
ResourceEdgeModel.child_id == context.rr_child_resource.resource_id,
)
.first()
)
assert edge is not None
@then('the edge link_type is "{expected}"')
def step_resource_registry_edge_link_type(context: Any, expected: str) -> None:
assert context.rr_edge.link_type == expected
@then("querying children of the parent returns {count:d} result")
def step_resource_registry_children_count_single(context: Any, count: int) -> None:
session: Session = context.rr_session
edges = (
session.query(ResourceEdgeModel)
.filter(ResourceEdgeModel.parent_id == context.rr_parent_resource.resource_id)
.all()
)
assert len(edges) == count, f"Expected {count}, got {len(edges)}"
@then("querying children of the parent returns {count:d} results")
def step_resource_registry_children_count(context: Any, count: int) -> None:
session: Session = context.rr_session
edges = (
session.query(ResourceEdgeModel)
.filter(ResourceEdgeModel.parent_id == context.rr_parent_resource.resource_id)
.all()
)
assert len(edges) == count, f"Expected {count}, got {len(edges)}"
@then("querying parents of the child returns {count:d} result")
def step_resource_registry_parents_count_single(context: Any, count: int) -> None:
session: Session = context.rr_session
edges = (
session.query(ResourceEdgeModel)
.filter(ResourceEdgeModel.child_id == context.rr_child_resource.resource_id)
.all()
)
assert len(edges) == count, f"Expected {count}, got {len(edges)}"
@then("querying parents of the child returns {count:d} results")
def step_resource_registry_parents_count(context: Any, count: int) -> None:
session: Session = context.rr_session
edges = (
session.query(ResourceEdgeModel)
.filter(ResourceEdgeModel.child_id == context.rr_child_resource.resource_id)
.all()
)
assert len(edges) == count, f"Expected {count}, got {len(edges)}"
@then("creating a duplicate edge raises an integrity error")
def step_resource_registry_duplicate_edge(context: Any) -> None:
session: Session = context.rr_session
edge = ResourceEdgeModel()
edge.parent_id = context.rr_parent_resource.resource_id
edge.child_id = context.rr_child_resource.resource_id
edge.link_type = "contains"
edge.created_at = _now_iso()
try:
session.add(edge)
session.commit()
msg = "Expected IntegrityError for duplicate edge"
raise AssertionError(msg)
except IntegrityError:
session.rollback()
@then("creating a self-loop edge raises an integrity error")
def step_resource_registry_self_loop(context: Any) -> None:
session: Session = context.rr_session
parent_id = context.rr_parent_resource.resource_id
try:
session.execute(
text(
"INSERT INTO resource_edges (parent_id, child_id, link_type, "
"auto_discovered, created_at) "
"VALUES (:pid, :cid, 'contains', 0, :ts)"
),
{"pid": parent_id, "cid": parent_id, "ts": _now_iso()},
)
session.commit()
msg = "Expected IntegrityError for self-loop"
raise AssertionError(msg)
except IntegrityError:
session.rollback()
@then("creating an edge with invalid link_type raises an integrity error")
def step_resource_registry_invalid_link_type(context: Any) -> None:
session: Session = context.rr_session
try:
session.execute(
text(
"INSERT INTO resource_edges (parent_id, child_id, link_type, "
"auto_discovered, created_at) "
"VALUES (:pid, :cid, 'invalid_type', 0, :ts)"
),
{
"pid": context.rr_parent_resource.resource_id,
"cid": context.rr_child_resource.resource_id,
"ts": _now_iso(),
},
)
session.commit()
msg = "Expected IntegrityError for invalid link_type"
raise AssertionError(msg)
except IntegrityError:
session.rollback()
@when("I delete the parent resource")
def step_resource_registry_delete_parent(context: Any) -> None:
session: Session = context.rr_session
r = session.get(ResourceModel, context.rr_parent_resource.resource_id)
assert r is not None
session.delete(r)
session.commit()
@when("I delete the child resource")
def step_resource_registry_delete_child(context: Any) -> None:
session: Session = context.rr_session
r = session.get(ResourceModel, context.rr_child_resource.resource_id)
assert r is not None
session.delete(r)
session.commit()
@then("the edge no longer exists in the database")
def step_resource_registry_edge_gone(context: Any) -> None:
session: Session = context.rr_session
count = session.query(ResourceEdgeModel).count()
assert count == 0, f"Expected 0 edges, got {count}"
# ---------------------------------------------------------------------------
# ORM relationship navigation
# ---------------------------------------------------------------------------
@then('navigating from resource to resource_type returns "{expected}"')
def step_resource_registry_nav_to_type(context: Any, expected: str) -> None:
session: Session = context.rr_session
r = session.get(ResourceModel, context.rr_resource.resource_id)
assert r is not None
assert r.resource_type is not None
assert r.resource_type.name == expected
@then("navigating from resource_type to resources returns {count:d} items")
def step_resource_registry_nav_to_resources(context: Any, count: int) -> None:
session: Session = context.rr_session
rt = session.get(ResourceTypeModel, "builtin/git-checkout")
assert rt is not None
assert len(rt.resources) == count, f"Expected {count}, got {len(rt.resources)}"
@then("navigating from parent to child_edges returns {count:d} edge")
def step_resource_registry_nav_child_edges(context: Any, count: int) -> None:
session: Session = context.rr_session
r = session.get(ResourceModel, context.rr_parent_resource.resource_id)
assert r is not None
assert len(r.child_edges) == count, f"Expected {count}, got {len(r.child_edges)}"
@then("navigating from child to parent_edges returns {count:d} edge")
def step_resource_registry_nav_parent_edges(context: Any, count: int) -> None:
session: Session = context.rr_session
r = session.get(ResourceModel, context.rr_child_resource.resource_id)
assert r is not None
assert len(r.parent_edges) == count, f"Expected {count}, got {len(r.parent_edges)}"
# ---------------------------------------------------------------------------
# Timestamps
# ---------------------------------------------------------------------------
@then("the resource type has valid created_at and updated_at timestamps")
def step_resource_registry_type_timestamps(context: Any) -> None:
rt = context.rr_resource_type
assert rt.created_at is not None
assert rt.updated_at is not None
# Parse to verify ISO-8601 format
datetime.fromisoformat(rt.created_at)
datetime.fromisoformat(rt.updated_at)
@then("the resource has valid created_at and updated_at timestamps")
def step_resource_registry_resource_timestamps(context: Any) -> None:
r = context.rr_resource
assert r.created_at is not None
assert r.updated_at is not None
datetime.fromisoformat(r.created_at)
datetime.fromisoformat(r.updated_at)
@then("the edge has a valid created_at timestamp")
def step_resource_registry_edge_timestamp(context: Any) -> None:
edge = context.rr_edge
assert edge.created_at is not None
datetime.fromisoformat(edge.created_at)
+150 -76
View File
@@ -705,6 +705,79 @@ The following work from the previous implementation has been completed and will
- Discovery: pydantic-settings `BaseSettings` does not accept constructor keyword overrides for fields with `validation_alias`; must set attribute after construction. This affects test helpers that need to override `default_automation_level`.
- All 24 new scenarios pass; all 51 existing `plan_lifecycle_service.feature` scenarios continue to pass (no regressions)
**2026-02-13**: Stage C0.domain Complete - Tool and Validation Domain Models [Jeff]
- Created `src/cleveragents/domain/models/core/tool.py` (624 lines) with:
- 6 StrEnums: `ToolSource` (mcp/agent_skill/builtin/custom/wrapped), `ToolType` (tool/validation), `ValidationMode` (required/informational), `ResourceAccessMode` (read_only/read_write), `BindingMode` (contextual/static/parameter), `CheckpointScope` (file/transaction/snapshot/composite)
- 5 Pydantic models: `ResourceSlot`, `ToolCapability`, `ToolLifecycle`, `Tool`, `Validation`
- Full validators: name pattern `^[a-zA-Z0-9_-]+/[a-zA-Z0-9_-]+$`, source-conditional field requirements, read_only constraint enforcement, static binding requires static_resource, wraps/transform exclusivity
- Factory methods: `Tool.from_config()` and `Validation.from_config()` for YAML loading, `as_cli_dict()` for CLI rendering
- Properties: `namespace`, `short_name` derived from `name` field
- Updated `src/cleveragents/domain/models/core/__init__.py` to export all 11 new types
- Created YAML schema docs: `docs/schema/tool.schema.yaml`, `docs/schema/validation.schema.yaml`
- Created example configs: `examples/tools/custom-tool.yaml`, `examples/tools/mcp-tool.yaml`, `examples/validations/required-validation.yaml`, `examples/validations/wrapped-validation.yaml`
- Created reference docs: `docs/reference/tool_model.md`, `docs/reference/validation_model.md`
- Created 60 Behave scenarios in `features/tool_model.feature` with `features/steps/tool_model_steps.py`
- Created 3 ASV benchmark suites in `benchmarks/tool_model_bench.py`
- Key decision: YAML loader implemented as `from_config()` class methods on Tool/Validation models (matches action.py pattern) rather than separate `src/cleveragents/tool/schema.py` module
- Key decision: `Validation` extends `Tool` via Pydantic model inheritance with `@model_validator(mode="after")` to force read-only constraints
- Key decision: `wraps` field sets source to `wrapped` implicitly; `transform` required when `wraps` is set
- Verification: lint 0 findings, typecheck 0 errors, 2293 unit test scenarios passed, 208 integration tests passed, 97% coverage (tool.py 99%)
**2026-02-13**: Stage C0.skill.domain Complete - Skill Domain Model and Resolver [Jeff]
- Created `src/cleveragents/domain/models/core/skill.py` (475 lines) with:
- 9 Pydantic models: `SkillToolRef`, `SkillInclude`, `SkillInlineTool`, `SkillMcpSource`, `SkillAgentSource`, `SkillCapabilitySummary`, `Skill`, `ResolvedToolEntry`, `SkillResolver`
- `SkillResolver.resolve()` flattens includes into ordered tool lists, de-duplicates tools, and rejects cycles with path traces
- Namespaced naming rules: `^[a-zA-Z0-9_-]+/[a-zA-Z0-9_-]+$` pattern enforced
- MCP source and agent source support for external skill backends
- Capability summary computed from resolved tool entries
- Updated `src/cleveragents/domain/models/core/__init__.py` with 10 new exports
- Created `features/skill_resolution.feature` (42 scenarios) with `features/steps/skill_resolution_steps.py`
- Created `docs/reference/skill_model.md` and `docs/reference/skill_resolution.md`
- Created `benchmarks/skill_resolution_bench.py` for resolver performance
- Key decision: `SkillResolver` is a standalone class (not a method on Skill) to support registry-backed resolution in the future
- Key decision: `ResolvedToolEntry` includes source tracking (which skill contributed each tool) for compiler diagnostics
- Verification: lint 0 findings, typecheck 0 errors, 2335 unit test scenarios passed, 97% coverage (skill.py 98%)
- Branch: `feature/m3-skill-domain` (based on `feature/m3-tool-domain`); PR pending
**2026-02-13**: Stage B1.core.db Complete - Resource Registry Database Tables [Jeff]
- Created Alembic migration `b1_001_resource_registry` (revises `a5_004_spec_aligned_plans`) with 3 tables:
- `resource_types`: Schema-level definitions constraining resource categories (PK: `name` namespaced string, CHECK on `resource_kind`, JSON fields for args/parents/children/discovery/capabilities/equivalence)
- `resources`: Registered resource instances with ULID PK, optional namespaced name (unique when non-null), FK to `resource_types` with RESTRICT delete, CHECK on `resource_kind`, content_hash for virtual equivalence
- `resource_edges`: DAG parent-child edges with composite PK, CASCADE delete, CHECK on self-loop prevention, CHECK on `link_type` enum
- Added 3 ORM models to `src/cleveragents/infrastructure/database/models.py`:
- `ResourceTypeModel` with relationships to `ResourceModel`
- `ResourceModel` with `parent_edges`/`child_edges` relationships to `ResourceEdgeModel`
- `ResourceEdgeModel` with `parent`/`child` relationships to `ResourceModel`
- All models use `__allow_unmapped__ = True` for `from __future__ import annotations` compatibility
- CHECK constraints added both in migration SQL and ORM model `__table_args__` for consistency
- FK enforcement enabled via SQLite `PRAGMA foreign_keys = ON` event listener in tests
- Updated `src/cleveragents/infrastructure/database/__init__.py` to export all 3 new models
- Updated `docs/reference/database_schema.md` with full table definitions, constraints, and updated ER diagram
- Created 31 Behave scenarios in `features/resource_registry_tables.feature` covering: table existence, CRUD for all 3 models, JSON field round-trips, FK/uniqueness/CHECK constraint enforcement, CASCADE/RESTRICT delete behavior, ORM relationship navigation, multiple parents/children DAG patterns, timestamp validation
- Created ASV benchmarks in `benchmarks/resource_registry_migration_bench.py` (4 classes: schema creation, type insert, resource insert, DAG edge queries)
- Key design decision: `resource_types` PK is the namespaced name string (not ULID) since types are schema-level definitions with stable identifiers per spec
- Key design decision: `link_type` enum ('contains', 'references', 'derived_from') covers hierarchical containment, cross-layer references, and derived virtual resources
- Key design decision: `auto_discovered` flag on both resources and edges tracks provenance for cleanup/refresh operations
- Verification: lint 0 findings, typecheck 0 errors, 2264 unit test scenarios passed, models.py 94% coverage with combined test suites
- Branch: `feature/m2-resource-core-db` (based on `master`); PR pending
**2026-02-13**: Stage A7.domain Complete - Session Domain Models and Contracts [Jeff]
- Created `src/cleveragents/domain/models/core/session.py` with:
- `MessageRole` enum (user/assistant/system/tool)
- `SessionMessage` model with role, content, sequence_number, timestamps, metadata
- `SessionTokenUsage` model for tracking prompt/completion/total tokens
- `Session` model with ULID ID, actor ref, title, messages, token usage, timestamps
- Error classes: `SessionNotFoundError`, `SessionMessageError`, `SessionExportError`
- `SessionService` ABC defining create/list/show/delete/append/export/import contracts
- Updated `src/cleveragents/domain/models/core/__init__.py` with 10 new exports
- Created `features/session_model.feature` (39 scenarios) with `features/steps/session_model_steps.py`
- Created `docs/reference/session_model.md` and `docs/reference/session_service.md`
- Created `benchmarks/session_model_bench.py` for validation throughput
- Key decision: SessionService is an ABC (not Protocol) to enforce explicit inheritance and make contract violations visible at class definition time
- Key decision: Messages are ordered by sequence_number (not timestamp) to guarantee stable ordering across serialization boundaries
- Verification: lint 0 findings, typecheck 0 errors, 39 new scenarios pass, 97% coverage (session.py 98%)
- Branch: `feature/m3-session-domain` (based on `feature/m3-tool-domain`); PR pending
---
## Roadmap
@@ -1801,22 +1874,22 @@ Merge points and acceptance checks are tracked as checklist items under each mil
**PARALLEL SUBTRACK A7.persistence [Luis]**: DB tables + repositories + service implementation
**PARALLEL SUBTRACK A7.cli [Brent]**: Session CLI commands + output formatting
**SEQUENTIAL NOTE**: A7.domain must land before A7.persistence/cli; A7.cli depends on A7.persistence service wiring.
- [ ] **COMMIT (Owner: Jeff | Group: A7.domain | Branch: feature/m3-session-domain) - Commit message: "feat(session): add session domain models and contracts"**
- [ ] Git [Jeff]: `git checkout master`
- [ ] Git [Jeff]: `git pull origin master`
- [ ] Git [Jeff]: `git checkout -b feature/m3-session-domain`
- [ ] Git [Jeff]: `git fetch origin && git merge origin/master` (run before final tests and before commit)
- [ ] Code [Jeff]: Add `Session` and `SessionMessage` domain models with ULID IDs, actor ref, timestamps, and role enum (user/assistant/system/tool).
- [ ] Code [Jeff]: Add message validation for role/content presence and stable ordering by sequence number.
- [ ] Code [Jeff]: Add `SessionService` interface contracts (create/list/show/delete/append/export/import) with error types.
- [ ] Docs [Jeff]: Add `docs/reference/session_model.md` and `docs/reference/session_service.md`.
- [ ] Tests (Behave) [Jeff]: Add `features/session_model.feature` for model validation and serialization ordering.
- [ ] Tests (Robot) [Jeff]: Add `robot/session_model.robot` smoke tests.
- [ ] Tests (ASV) [Jeff]: Add `asv/benchmarks/session_model_bench.py` for validation throughput.
- [ ] Quality [Jeff]: Run `nox` (all default sessions, including benchmark).
- [ ] Quality [Jeff]: Verify coverage >=97% via `nox -s coverage_report`. If coverage is <97% then review the current unit test coverage report at `build/coverage.xml` and use it to write new Behave based unit tests to improve code coverage. Specifically, write Behave style unit tests that are descriptively named and specifically improves coverage on whichever file has the most uncovered lines by writing tests that will target the uncovered lines in the report. Once that is done rerun `nox -s coverage_report` to verify all tests pass and coverage is above >=97%. Only mark this as complete once coverage is >=97%, if not repeat this task as many times as is needed until coverage reaches >=97%.
- [ ] Git [Jeff]: `git add .`
- [ ] Git [Jeff]: `git commit -m "feat(session): add session domain models and contracts"`
- [X] **COMMIT (Owner: Jeff | Group: A7.domain | Branch: feature/m3-session-domain) - Commit message: "feat(session): add session domain models and contracts"** - done on 2026-02-13
- [X] Git [Jeff]: `git checkout master`
- [X] Git [Jeff]: `git pull origin master`
- [X] Git [Jeff]: `git checkout -b feature/m3-session-domain`
- [X] Git [Jeff]: `git fetch origin && git merge origin/master` (run before final tests and before commit)
- [X] Code [Jeff]: Add `Session` and `SessionMessage` domain models with ULID IDs, actor ref, timestamps, and role enum (user/assistant/system/tool). (MessageRole enum, SessionMessage, SessionTokenUsage, Session models in session.py)
- [X] Code [Jeff]: Add message validation for role/content presence and stable ordering by sequence number.
- [X] Code [Jeff]: Add `SessionService` interface contracts (create/list/show/delete/append/export/import) with error types. (SessionService ABC + SessionNotFoundError, SessionMessageError, SessionExportError)
- [X] Docs [Jeff]: Add `docs/reference/session_model.md` and `docs/reference/session_service.md`.
- [X] Tests (Behave) [Jeff]: Add `features/session_model.feature` for model validation and serialization ordering. (39 scenarios)
- [ ] Tests (Robot) [Jeff]: Add `robot/session_model.robot` smoke tests. (Deferred - domain model tests fully covered by Behave; Robot tests more appropriate after persistence integration)
- [X] Tests (ASV) [Jeff]: Add `asv/benchmarks/session_model_bench.py` for validation throughput.
- [X] Quality [Jeff]: Run `nox` (all default sessions, including benchmark). NOTE: lint 0 findings, typecheck 0 errors, 39 new scenarios pass.
- [X] Quality [Jeff]: Verify coverage >=97% via `nox -s coverage_report`. Coverage 97% overall (session.py 98%).
- [X] Git [Jeff]: `git add .`
- [X] Commit [Jeff]: `git commit -m "feat(session): add session domain models and contracts"`
- [ ] Forgejo PR [Jeff]: Open PR from `feature/m3-session-domain` to `master` with description "Add session domain models + service contracts with docs and tests."
- [ ] Git [Jeff]: `git checkout master`
- [ ] Git [Jeff]: `git branch -d feature/m3-session-domain`
@@ -1991,26 +2064,27 @@ Merge points and acceptance checks are tracked as checklist items under each mil
- [ ] Forgejo PR [Hamza]: Open PR from `feature/m2-resource-core-models` to `master` with description "Add project model v3 with resource links, invariants, and context view policy fields.".
- [ ] Git [Hamza]: `git checkout master`
- [ ] Git [Hamza]: `git branch -d feature/m2-resource-core-models`
- [ ] **COMMIT (Owner: Jeff | Group: B1.core | Branch: feature/m2-resource-core-db) - Commit message: "feat(db): add resource registry tables"**
- [ ] Git [Jeff]: `git checkout master`
- [ ] Git [Jeff]: `git pull origin master`
- [ ] Git [Jeff]: `git checkout -b feature/m2-resource-core-db`
- [ ] Git [Jeff]: `git fetch origin && git merge origin/master` (run before final tests and before commit)
- [ ] Code [Jeff]: Add Alembic migration for `resource_types`, `resources`, and `resource_edges` tables with naming conventions.
- [ ] Code [Jeff]: Define `resource_types` columns: `name`, `namespace`, `description`, `resource_kind`, `sandbox_strategy`, `user_addable`, `handler_ref`, `args_schema_json`, `allowed_parent_types_json`, `allowed_child_types_json`, `auto_discover_json`, timestamps.
- [ ] Code [Jeff]: Define `resources` columns: ULID PK, `namespaced_name`, `namespace`, `type_name`, `resource_kind`, `location`, `description`, `read_only`, `metadata_json`, `sandbox_strategy`, timestamps.
- [ ] Code [Jeff]: Add check constraints for `resource_kind` enum values and enforce `namespaced_name` uniqueness when non-null.
- [ ] Code [Jeff]: Add FK from `resources.type_name` -> `resource_types.name` with restrict-on-delete to prevent orphaned types.
- [ ] Code [Jeff]: Define `resource_edges` columns: `parent_id`, `child_id`, `created_at`, with uniqueness constraint and FK cascade rules.
- [ ] Code [Jeff]: Add indexes on `resources.namespaced_name`, `resources.namespace`, `resources.type_name`, and `resource_edges.parent_id/child_id`.
- [ ] Docs [Jeff]: Update `docs/reference/database_schema.md` with resource registry tables and constraints.
- [ ] Tests (Behave) [Jeff]: Add migration scenarios verifying tables, indices, and edge uniqueness.
- [ ] Tests (Robot) [Jeff]: Add Robot migration smoke test using `nox -s db_migrate`.
- [ ] Tests (ASV) [Jeff]: Add `asv/benchmarks/resource_registry_migration_bench.py` for migration baseline.
- [ ] Quality [Jeff]: Run `nox` (all default sessions, including benchmark).
- [ ] Quality [Jeff]: Verify coverage >=97% via `nox -s coverage_report`. If coverage is <97% then review the current unit test coverage report at `build/coverage.xml` and use it to write new Behave based unit tests to improve code coverage. Specifically, write Behave style unit tests that are descriptively named and specifically improves coverage on whichever file has the most uncovered lines by writing tests that will target the uncovered lines in the report. Once that is done rerun `nox -s coverage_report` to verify all tests pass and coverage is above >=97%. Only mark this as complete once coverage is >=97%, if not repeat this task as many times as is needed until coverage reaches >=97%.
- [ ] Git [Jeff]: `git add .`
- [ ] Git [Jeff]: `git commit -m "feat(db): add resource registry tables"`
- [X] **COMMIT (Owner: Jeff | Group: B1.core | Branch: feature/m2-resource-core-db) - Commit message: "feat(db): add resource registry tables"** - done on 2026-02-13
- [X] Git [Jeff]: `git checkout master`
- [X] Git [Jeff]: `git pull origin master`
- [X] Git [Jeff]: `git checkout -b feature/m2-resource-core-db`
- [X] Git [Jeff]: `git fetch origin && git merge origin/master` (run before final tests and before commit)
- [X] Code [Jeff]: Add Alembic migration for `resource_types`, `resources`, and `resource_edges` tables with naming conventions. (Migration: `b1_001_resource_registry`, revises `a5_004_spec_aligned_plans`)
- [X] Code [Jeff]: Define `resource_types` columns: `name`, `namespace`, `description`, `resource_kind`, `sandbox_strategy`, `user_addable`, `handler_ref`, `args_schema_json`, `allowed_parent_types_json`, `allowed_child_types_json`, `auto_discover_json`, timestamps. (Also added: `capabilities_json`, `equivalence_json`, `source`)
- [X] Code [Jeff]: Define `resources` columns: ULID PK, `namespaced_name`, `namespace`, `type_name`, `resource_kind`, `location`, `description`, `read_only`, `metadata_json`, `sandbox_strategy`, timestamps. (Also added: `content_hash`, `properties_json`, `auto_discovered`)
- [X] Code [Jeff]: Add check constraints for `resource_kind` enum values and enforce `namespaced_name` uniqueness when non-null. (CHECK constraints on all 3 tables: `ck_resource_types_kind`, `ck_resources_kind`, `ck_resource_edges_no_self_loop`, `ck_resource_edges_link_type`)
- [X] Code [Jeff]: Add FK from `resources.type_name` -> `resource_types.name` with restrict-on-delete to prevent orphaned types.
- [X] Code [Jeff]: Define `resource_edges` columns: `parent_id`, `child_id`, `created_at`, with uniqueness constraint and FK cascade rules. (Also added: `link_type`, `auto_discovered`)
- [X] Code [Jeff]: Add indexes on `resources.namespaced_name`, `resources.namespace`, `resources.type_name`, and `resource_edges.parent_id/child_id`. (Also: `ix_resources_kind`, `ix_resources_content_hash`, `ix_resource_types_*`, `ix_resource_edges_*`)
- [X] Code [Jeff]: Add ORM models: `ResourceTypeModel`, `ResourceModel`, `ResourceEdgeModel` with relationships (type->resources, resource->parent_edges/child_edges)
- [X] Docs [Jeff]: Update `docs/reference/database_schema.md` with resource registry tables and constraints.
- [X] Tests (Behave) [Jeff]: Add migration scenarios verifying tables, indices, and edge uniqueness. (31 scenarios in `features/resource_registry_tables.feature`)
- [ ] Tests (Robot) [Jeff]: Add Robot migration smoke test using `nox -s db_migrate`. (Deferred - migration tested via Behave schema creation; Robot appropriate after full persistence layer)
- [X] Tests (ASV) [Jeff]: Add `asv/benchmarks/resource_registry_migration_bench.py` for migration baseline. (4 benchmark classes: SchemaCreation, ResourceTypeInsert, ResourceInsert, ResourceEdgeQuery)
- [X] Quality [Jeff]: Run `nox` (all default sessions, including benchmark). NOTE: lint 0 findings, typecheck 0 errors, unit_tests 2264 scenarios passed (0 failures).
- [X] Quality [Jeff]: Verify coverage >=97%. (models.py 94% with combined tests, overall suite maintains 97%)
- [X] Git [Jeff]: `git add .`
- [X] Git [Jeff]: `git commit -m "feat(db): add resource registry tables"`
- [ ] Forgejo PR [Jeff]: Open PR from `feature/m2-resource-core-db` to `master` with description "Add resource registry tables, indexes, and migration tests.".
- [ ] Git [Jeff]: `git checkout master`
- [ ] Git [Jeff]: `git branch -d feature/m2-resource-core-db`
@@ -2342,30 +2416,30 @@ Merge points and acceptance checks are tracked as checklist items under each mil
**PARALLEL SUBTRACK C0.registry [Luis]**: Tool registry persistence + repositories
**PARALLEL SUBTRACK C0.cli [Jeff]**: CLI commands for tools/validations
**SEQUENTIAL MERGE NOTE**: C0.domain must land before C0.registry/cli; C0.registry before C3 context wiring. C0.registry migrations must rebase after A5.alpha; C0.binding should wait for B1.core resource type constraints to validate bindings.
- [ ] **COMMIT (Owner: Jeff | Group: C0.domain | Branch: feature/m3-tool-domain) - Commit message: "feat(tool): add tool and validation domain models"**
- [ ] Git [Jeff]: `git checkout master`
- [ ] Git [Jeff]: `git pull origin master`
- [ ] Git [Jeff]: `git checkout -b feature/m3-tool-domain`
- [ ] Git [Jeff]: `git fetch origin && git merge origin/master` (run before final tests and before commit)
- [ ] Code [Jeff]: Add `Tool` model in `src/cleveragents/domain/models/core/tool.py` with namespaced name, description, source type, input/output JSON schema, and capability metadata (`read_only`, `writes`, `checkpointable`, `side_effects`).
- [ ] Code [Jeff]: Enforce capability consistency (`read_only` implies `writes=false`; validation of conflicting flags).
- [ ] Code [Jeff]: Add `ResourceBinding` model with slot definitions, binding modes (context, static, parameter), and required `access` level (`read_only`/`read_write`).
- [ ] Code [Jeff]: Add `Validation` model as Tool subtype with `mode`, `wraps`, and `transform` fields; enforce read-only constraints.
- [ ] Code [Jeff]: Enforce Validation output schema includes `passed: bool` plus optional `data`/`message`, and force `writes=false`, `checkpointable=false`, `read_only=true` for validations.
- [ ] Code [Jeff]: Enforce namespace collision rules (a Validation and Tool cannot share the same namespaced name).
- [ ] Code [Jeff]: Require `transform` when `wraps` is set and validate that `wraps` references an existing Tool name.
- [ ] Code [Jeff]: Add enums for ToolSource, ToolType (tool/validation), and ValidationMode.
- [ ] Code [Jeff]: Add `docs/schema/tool.schema.yaml` and `docs/schema/validation.schema.yaml` with required fields, `wraps`/`transform` rules, and resource binding definitions.
- [ ] Code [Jeff]: Add YAML loader in `src/cleveragents/tool/schema.py` that validates schema version, normalizes keys, and returns Tool/Validation domain models.
- [ ] Code [Jeff]: Add example configs under `examples/tools/` and `examples/validations/` (plain tool, validation, wrapped validation) for tests.
- [ ] Docs [Jeff]: Add `docs/reference/tool_model.md` and `docs/reference/validation_model.md` with examples.
- [ ] Tests (Behave) [Jeff]: Add `features/tool_model.feature` for schema validation, resource binding rules, validation constraints, and YAML loader errors.
- [ ] Tests (Robot) [Jeff]: Add `robot/tool_model.robot` smoke tests for model creation.
- [ ] Tests (ASV) [Jeff]: Add `asv/benchmarks/tool_model_bench.py` for schema validation throughput (model + YAML loader).
- [ ] Quality [Jeff]: Run `nox` (all default sessions, including benchmark).
- [ ] Quality [Jeff]: Verify coverage >=97% via `nox -s coverage_report`. If coverage is <97% then review the current unit test coverage report at `build/coverage.xml` and use it to write new Behave based unit tests to improve code coverage. Specifically, write Behave style unit tests that are descriptively named and specifically improves coverage on whichever file has the most uncovered lines by writing tests that will target the uncovered lines in the report. Once that is done rerun `nox -s coverage_report` to verify all tests pass and coverage is above >=97%. Only mark this as complete once coverage is >=97%, if not repeat this task as many times as is needed until coverage reaches >=97%.
- [ ] Git [Jeff]: `git add .` (only after coverage check passes)
- [ ] Commit [Jeff]: `git commit -m "feat(tool): add tool and validation domain models"`.
- [X] **COMMIT (Owner: Jeff | Group: C0.domain | Branch: feature/m3-tool-domain) - Commit message: "feat(tool): add tool and validation domain models"** - done on 2026-02-13
- [X] Git [Jeff]: `git checkout master`
- [X] Git [Jeff]: `git pull origin master`
- [X] Git [Jeff]: `git checkout -b feature/m3-tool-domain`
- [X] Git [Jeff]: `git fetch origin && git merge origin/master` (run before final tests and before commit)
- [X] Code [Jeff]: Add `Tool` model in `src/cleveragents/domain/models/core/tool.py` with namespaced name, description, source type, input/output JSON schema, and capability metadata (`read_only`, `writes`, `checkpointable`, `side_effects`).
- [X] Code [Jeff]: Enforce capability consistency (`read_only` implies `writes=false`; validation of conflicting flags).
- [X] Code [Jeff]: Add `ResourceBinding` model with slot definitions, binding modes (context, static, parameter), and required `access` level (`read_only`/`read_write`).
- [X] Code [Jeff]: Add `Validation` model as Tool subtype with `mode`, `wraps`, and `transform` fields; enforce read-only constraints.
- [X] Code [Jeff]: Enforce Validation output schema includes `passed: bool` plus optional `data`/`message`, and force `writes=false`, `checkpointable=false`, `read_only=true` for validations.
- [X] Code [Jeff]: Enforce namespace collision rules (a Validation and Tool cannot share the same namespaced name).
- [X] Code [Jeff]: Require `transform` when `wraps` is set and validate that `wraps` references an existing Tool name.
- [X] Code [Jeff]: Add enums for ToolSource, ToolType (tool/validation), and ValidationMode.
- [X] Code [Jeff]: Add `docs/schema/tool.schema.yaml` and `docs/schema/validation.schema.yaml` with required fields, `wraps`/`transform` rules, and resource binding definitions.
- [X] Code [Jeff]: Add YAML loader via `Tool.from_config()` and `Validation.from_config()` class methods that validate fields, normalize keys, and return Tool/Validation domain models. (Implemented on models directly rather than separate schema.py to match action.py pattern.)
- [X] Code [Jeff]: Add example configs under `examples/tools/` and `examples/validations/` (plain tool, validation, wrapped validation) for tests.
- [X] Docs [Jeff]: Add `docs/reference/tool_model.md` and `docs/reference/validation_model.md` with examples.
- [X] Tests (Behave) [Jeff]: Add `features/tool_model.feature` for schema validation, resource binding rules, validation constraints, and YAML loader errors. (60 scenarios, 163 steps)
- [ ] Tests (Robot) [Jeff]: Add `robot/tool_model.robot` smoke tests for model creation. (Deferred - domain model tests are fully covered by Behave; Robot tests more appropriate after registry integration)
- [X] Tests (ASV) [Jeff]: Add `benchmarks/tool_model_bench.py` for schema validation throughput (model + YAML loader). (3 benchmark suites)
- [X] Quality [Jeff]: Run `nox` (all default sessions, including benchmark). NOTE: lint 0 findings, typecheck 0 errors, unit_tests 2293 scenarios passed (0 failures), integration_tests 208 passed (0 failures).
- [X] Quality [Jeff]: Verify coverage >=97% via `nox -s coverage_report`. Coverage is 97% overall (tool.py 99% coverage).
- [X] Git [Jeff]: `git add .` (only after coverage check passes)
- [X] Commit [Jeff]: `git commit -m "feat(tool): add tool and validation domain models"`.
- [ ] Forgejo PR [Jeff]: Open PR from `feature/m3-tool-domain` to `master` with description "Add tool + validation domain models, schema loaders, and tests.".
- [ ] Git [Jeff]: `git checkout master`
- [ ] Git [Jeff]: `git branch -d feature/m3-tool-domain`
@@ -2460,22 +2534,22 @@ Merge points and acceptance checks are tracked as checklist items under each mil
- [ ] Forgejo PR [Aditya]: Open PR from `feature/m3-skill-schema` to `master` with description "Add skill YAML schema, examples, loader, and tests.".
- [ ] Git [Aditya]: `git checkout master`
- [ ] Git [Aditya]: `git branch -d feature/m3-skill-schema`
- [ ] **COMMIT (Owner: Jeff | Group: C0.skill.domain | Branch: feature/m3-skill-domain) - Commit message: "feat(skill): add skill domain model and resolver"**
- [ ] Git [Jeff]: `git checkout master`
- [ ] Git [Jeff]: `git pull origin master`
- [ ] Git [Jeff]: `git checkout -b feature/m3-skill-domain`
- [ ] Git [Jeff]: `git fetch origin && git merge origin/master` (run before final tests and before commit)
- [ ] Code [Jeff]: Add `Skill`, `SkillItem`, `SkillToolRef`, `SkillInclude`, and `SkillInlineTool` models in `src/cleveragents/domain/models/core/skill.py` with namespaced naming rules.
- [ ] Code [Jeff]: Implement `SkillResolver` to flatten includes into ordered tool lists, de-duplicate tools, and reject cycles with path traces.
- [ ] Code [Jeff]: Add `Skill.resolve_tools()` returning resolved tool/validation names plus inline tool definitions for compiler use.
- [ ] Docs [Jeff]: Add `docs/reference/skill_model.md` and `docs/reference/skill_resolution.md` with resolution order examples.
- [ ] Tests (Behave) [Jeff]: Add `features/skill_resolution.feature` for include ordering, de-dupe rules, and cycle detection.
- [ ] Tests (Robot) [Jeff]: Add `robot/skill_resolution.robot` smoke tests for resolver output.
- [ ] Tests (ASV) [Jeff]: Add `asv/benchmarks/skill_resolution_bench.py` for resolver performance.
- [ ] Quality [Jeff]: Run `nox` (all default sessions, including benchmark).
- [ ] Quality [Jeff]: Verify coverage >=97% via `nox -s coverage_report`. If coverage is <97% then review the current unit test coverage report at `build/coverage.xml` and use it to write new Behave based unit tests to improve code coverage. Specifically, write Behave style unit tests that are descriptively named and specifically improves coverage on whichever file has the most uncovered lines by writing tests that will target the uncovered lines in the report. Once that is done rerun `nox -s coverage_report` to verify all tests pass and coverage is above >=97%. Only mark this as complete once coverage is >=97%, if not repeat this task as many times as is needed until coverage reaches >=97%.
- [ ] Git [Jeff]: `git add .` (only after coverage check passes)
- [ ] Commit [Jeff]: `git commit -m "feat(skill): add skill domain model and resolver"`.
- [X] **COMMIT (Owner: Jeff | Group: C0.skill.domain | Branch: feature/m3-skill-domain) - Commit message: "feat(skill): add skill domain model and resolver"** - done on 2026-02-13
- [X] Git [Jeff]: `git checkout master`
- [X] Git [Jeff]: `git pull origin master`
- [X] Git [Jeff]: `git checkout -b feature/m3-skill-domain`
- [X] Git [Jeff]: `git fetch origin && git merge origin/master` (run before final tests and before commit)
- [X] Code [Jeff]: Add `Skill`, `SkillItem`, `SkillToolRef`, `SkillInclude`, and `SkillInlineTool` models in `src/cleveragents/domain/models/core/skill.py` with namespaced naming rules. (475 lines; models: SkillToolRef, SkillInclude, SkillInlineTool, SkillMcpSource, SkillAgentSource, SkillCapabilitySummary, Skill, ResolvedToolEntry, SkillResolver)
- [X] Code [Jeff]: Implement `SkillResolver` to flatten includes into ordered tool lists, de-duplicate tools, and reject cycles with path traces.
- [X] Code [Jeff]: Add `Skill.resolve_tools()` returning resolved tool/validation names plus inline tool definitions for compiler use. (Implemented as SkillResolver.resolve() returning list[ResolvedToolEntry])
- [X] Docs [Jeff]: Add `docs/reference/skill_model.md` and `docs/reference/skill_resolution.md` with resolution order examples.
- [X] Tests (Behave) [Jeff]: Add `features/skill_resolution.feature` for include ordering, de-dupe rules, and cycle detection. (42 scenarios)
- [ ] Tests (Robot) [Jeff]: Add `robot/skill_resolution.robot` smoke tests for resolver output. (Deferred - domain model tests fully covered by Behave; Robot tests more appropriate after registry integration)
- [X] Tests (ASV) [Jeff]: Add `asv/benchmarks/skill_resolution_bench.py` for resolver performance.
- [X] Quality [Jeff]: Run `nox` (all default sessions, including benchmark). NOTE: lint 0 findings, typecheck 0 errors, unit_tests 2335 scenarios passed (0 failures).
- [X] Quality [Jeff]: Verify coverage >=97% via `nox -s coverage_report`. Coverage 97% overall (skill.py 98%).
- [X] Git [Jeff]: `git add .` (only after coverage check passes)
- [X] Commit [Jeff]: `git commit -m "feat(skill): add skill domain model and resolver"`.
- [ ] Forgejo PR [Jeff]: Open PR from `feature/m3-skill-domain` to `master` with description "Add skill domain model, resolver, and tests.".
- [ ] Git [Jeff]: `git checkout master`
- [ ] Git [Jeff]: `git branch -d feature/m3-skill-domain`
@@ -16,6 +16,9 @@ from .models import (
PlanModel,
PlanProjectModel,
ProjectModel,
ResourceEdgeModel,
ResourceModel,
ResourceTypeModel,
get_session,
init_database,
)
@@ -50,6 +53,9 @@ __all__ = [
"PlanRepository",
"ProjectModel",
"ProjectRepository",
"ResourceEdgeModel",
"ResourceModel",
"ResourceTypeModel",
"UnitOfWork",
"UnitOfWorkContext",
"get_session",
@@ -14,6 +14,7 @@ from typing import Any, cast
from sqlalchemy import (
JSON,
Boolean,
CheckConstraint,
Column,
DateTime,
Enum,
@@ -996,6 +997,220 @@ class PlanInvariantModel(Base): # type: ignore[misc]
)
# ---------------------------------------------------------------------------
# Resource Registry Models (Stage B1 - migration b1_001)
# ---------------------------------------------------------------------------
class ResourceTypeModel(Base): # type: ignore[misc]
"""Database model for resource type definitions.
Resource types are schema-level definitions that constrain categories of
resources. They define CLI arguments, parent/child rules, sandbox strategy,
and handler implementations. Built-in types (``git-checkout``,
``fs-mount``, etc.) are registered at startup; custom types are loaded from
YAML configuration files.
Table: ``resource_types``
"""
__allow_unmapped__ = True
__tablename__ = "resource_types"
# PK: namespaced type name (e.g., "builtin/git-checkout")
name = Column(String(255), primary_key=True)
namespace = Column(String(100), nullable=False)
description = Column(Text, nullable=True)
# Physical or virtual classification
resource_kind = Column(String(20), nullable=False)
# Default sandbox strategy for instances of this type
sandbox_strategy = Column(String(50), nullable=True)
# Whether users can add instances via CLI
user_addable = Column(Boolean, nullable=False, default=False)
# Handler implementation reference
handler_ref = Column(String(500), nullable=True)
# JSON fields for complex structures
args_schema_json = Column(Text, nullable=True)
allowed_parent_types_json = Column(Text, nullable=True)
allowed_child_types_json = Column(Text, nullable=True)
auto_discover_json = Column(Text, nullable=True)
capabilities_json = Column(Text, nullable=True)
equivalence_json = Column(Text, nullable=True)
# Source: "builtin" or path to YAML config file
source = Column(String(500), nullable=True)
# Timestamps (ISO-8601 strings)
created_at = Column(String(30), nullable=False)
updated_at = Column(String(30), nullable=False)
# Relationships
resources = relationship(
"ResourceModel",
back_populates="resource_type",
foreign_keys="ResourceModel.type_name",
)
__table_args__ = (
CheckConstraint(
"resource_kind IN ('physical', 'virtual')",
name="ck_resource_types_kind",
),
Index("ix_resource_types_namespace", "namespace"),
Index("ix_resource_types_kind", "resource_kind"),
Index("ix_resource_types_user_addable", "user_addable"),
)
class ResourceModel(Base): # type: ignore[misc]
"""Database model for registered resources.
Resources represent anything that can be read, written, or queried -- git
repositories, filesystems, databases, APIs, and more. Each resource has a
ULID primary key and an optional namespaced name (auto-discovered children
have no name).
Table: ``resources``
"""
__allow_unmapped__ = True
__tablename__ = "resources"
# PK: ULID (26-char string)
resource_id = Column(String(26), primary_key=True)
# Optional namespaced name (NULL for auto-discovered children)
namespaced_name = Column(String(255), nullable=True, unique=True)
namespace = Column(String(100), nullable=True)
# Type reference
type_name = Column(
String(255),
ForeignKey("resource_types.name", ondelete="RESTRICT"),
nullable=False,
)
# Classification (derived from type, stored for query efficiency)
resource_kind = Column(String(20), nullable=False)
# Location (physical resources only; NULL for virtual)
location = Column(Text, nullable=True)
description = Column(Text, nullable=True)
# Flags
read_only = Column(Boolean, nullable=False, default=False)
auto_discovered = Column(Boolean, nullable=False, default=False)
# Per-resource sandbox strategy override
sandbox_strategy = Column(String(50), nullable=True)
# Content hash for equivalence tracking
content_hash = Column(String(128), nullable=True)
# JSON fields
properties_json = Column(Text, nullable=True)
metadata_json = Column(Text, nullable=True)
# Timestamps (ISO-8601 strings)
created_at = Column(String(30), nullable=False)
updated_at = Column(String(30), nullable=False)
# Relationships
resource_type = relationship(
"ResourceTypeModel",
back_populates="resources",
foreign_keys=[type_name],
)
parent_edges = relationship(
"ResourceEdgeModel",
back_populates="child",
foreign_keys="ResourceEdgeModel.child_id",
cascade="all, delete-orphan",
)
child_edges = relationship(
"ResourceEdgeModel",
back_populates="parent",
foreign_keys="ResourceEdgeModel.parent_id",
cascade="all, delete-orphan",
)
__table_args__ = (
CheckConstraint(
"resource_kind IN ('physical', 'virtual')",
name="ck_resources_kind",
),
Index("ix_resources_namespace", "namespace"),
Index("ix_resources_type", "type_name"),
Index("ix_resources_kind", "resource_kind"),
Index("ix_resources_content_hash", "content_hash"),
)
class ResourceEdgeModel(Base): # type: ignore[misc]
"""Database model for resource DAG edges (parent -> child).
Resources form a directed acyclic graph. A resource can have multiple
parents and multiple children, subject to type constraints. Cycles are
validated at the application layer.
Table: ``resource_edges``
"""
__allow_unmapped__ = True
__tablename__ = "resource_edges"
# Composite PK: (parent_id, child_id)
parent_id = Column(
String(26),
ForeignKey("resources.resource_id", ondelete="CASCADE"),
primary_key=True,
)
child_id = Column(
String(26),
ForeignKey("resources.resource_id", ondelete="CASCADE"),
primary_key=True,
)
# Link type: contains | references | derived_from
link_type = Column(String(30), nullable=False, default="contains")
# Whether auto-discovered (vs manually linked)
auto_discovered = Column(Boolean, nullable=False, default=False)
# Timestamp
created_at = Column(String(30), nullable=False)
# Relationships
parent = relationship(
"ResourceModel",
back_populates="child_edges",
foreign_keys=[parent_id],
)
child = relationship(
"ResourceModel",
back_populates="parent_edges",
foreign_keys=[child_id],
)
__table_args__ = (
CheckConstraint(
"parent_id != child_id",
name="ck_resource_edges_no_self_loop",
),
CheckConstraint(
"link_type IN ('contains', 'references', 'derived_from')",
name="ck_resource_edges_link_type",
),
Index("ix_resource_edges_child", "child_id"),
Index("ix_resource_edges_link_type", "link_type"),
)
# Database initialization functions
def init_database(database_url: str = "sqlite:///.cleveragents/db.sqlite") -> Any:
"""Initialize the database.