feat(resource): add DAG linking and discovery

alembic/versions/b1_001_resource_links.py

@@ -0,0 +1,87 @@
+"""Add resource_links table for validated DAG parent-child links.
+
+This migration creates the ``resource_links`` table that stores
+validated parent-child relationships between resources.  Unlike the
+``resource_edges`` table (which stores raw DAG edges with link-type
+metadata), ``resource_links`` records relationships that have passed
+cycle detection and type compatibility checks.
+
+Revision ID: b1_001_resource_links
+Revises: b1_001_resource_registry
+Create Date: 2026-02-17 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_links"
+down_revision: str | Sequence[str] | None = "b1_001_resource_registry"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    """Create the resource_links table."""
+    op.create_table(
+        "resource_links",
+        sa.Column(
+            "parent_id",
+            sa.String(26),
+            sa.ForeignKey(
+                "resources.resource_id",
+                ondelete="CASCADE",
+                name="fk_resource_links_parent",
+            ),
+            nullable=False,
+        ),
+        sa.Column(
+            "child_id",
+            sa.String(26),
+            sa.ForeignKey(
+                "resources.resource_id",
+                ondelete="CASCADE",
+                name="fk_resource_links_child",
+            ),
+            nullable=False,
+        ),
+        sa.Column(
+            "created_at",
+            sa.String(30),
+            nullable=False,
+        ),
+        sa.PrimaryKeyConstraint("parent_id", "child_id"),
+        sa.CheckConstraint(
+            "parent_id != child_id",
+            name="ck_resource_links_no_self_loop",
+        ),
+    )
+
+    op.create_index(
+        "ix_resource_links_child",
+        "resource_links",
+        ["child_id"],
+        unique=False,
+    )
+    op.create_index(
+        "ix_resource_links_parent",
+        "resource_links",
+        ["parent_id"],
+        unique=False,
+    )
+
+
+def downgrade() -> None:
+    """Drop the resource_links table."""
+    op.drop_index(
+        "ix_resource_links_parent",
+        table_name="resource_links",
+    )
+    op.drop_index(
+        "ix_resource_links_child",
+        table_name="resource_links",
+    )
+    op.drop_table("resource_links")

benchmarks/resource_dag_bench.py

@@ -0,0 +1,235 @@
+"""ASV benchmarks for resource DAG link and auto_discover."""
+
+from __future__ import annotations
+
+import json
+from datetime import UTC, datetime
+from typing import Any
+
+
+def _setup_db() -> Any:
+    """Create in-memory database and return session factory."""
+    from sqlalchemy import create_engine, event
+    from sqlalchemy.orm import sessionmaker
+
+    from cleveragents.infrastructure.database.models import Base
+
+    engine = create_engine("sqlite:///:memory:")
+
+    @event.listens_for(engine, "connect")
+    def _fk(conn: Any, _rec: Any) -> None:
+        conn.cursor().execute("PRAGMA foreign_keys=ON")
+
+    Base.metadata.create_all(engine)
+    return sessionmaker(bind=engine)
+
+
+_CB32 = "0123456789ABCDEFGHJKMNPQRSTVWXYZ"
+_BENCH_CTR = 0
+
+
+def _bench_ulid() -> str:
+    global _BENCH_CTR
+    _BENCH_CTR += 1
+    n = _BENCH_CTR
+    suffix = ""
+    for _ in range(8):
+        suffix = _CB32[n % 32] + suffix
+        n //= 32
+    return f"01HDAGBNCH0AQDYTR4B{suffix:>7s}"[:26]
+
+
+def _seed_types(
+    rt_repo: Any,
+    parent_name: str = "bench/dag-parent",
+    child_name: str = "bench/dag-child",
+) -> None:
+    """Seed parent and child resource types."""
+    from cleveragents.domain.models.core.resource_type import (
+        ResourceKind,
+        ResourceTypeSpec,
+        SandboxStrategy,
+    )
+
+    parent_spec = ResourceTypeSpec(
+        name=parent_name,
+        description="Bench parent type",
+        resource_kind=ResourceKind.PHYSICAL,
+        sandbox_strategy=SandboxStrategy.NONE,
+        user_addable=True,
+        cli_args=[],
+        parent_types=[],
+        child_types=[child_name],
+        auto_discovery={
+            "enabled": True,
+            "rules": [{"type": child_name, "pattern": "*"}],
+        },
+        equivalence=None,
+        handler=None,
+        capabilities={
+            "read": True,
+            "write": True,
+            "sandbox": True,
+            "checkpoint": False,
+        },
+        built_in=False,
+    )
+    child_spec = ResourceTypeSpec(
+        name=child_name,
+        description="Bench child type",
+        resource_kind=ResourceKind.PHYSICAL,
+        sandbox_strategy=SandboxStrategy.NONE,
+        user_addable=True,
+        cli_args=[],
+        parent_types=[],
+        child_types=[],
+        auto_discovery=None,
+        equivalence=None,
+        handler=None,
+        capabilities={
+            "read": True,
+            "write": True,
+            "sandbox": True,
+            "checkpoint": False,
+        },
+        built_in=False,
+    )
+    rt_repo.create(parent_spec)
+    rt_repo.create(child_spec)
+
+
+def _make_resource(
+    res_repo: Any,
+    type_name: str,
+) -> str:
+    """Create a resource and return its ID."""
+    from cleveragents.domain.models.core.resource import (
+        PhysVirt,
+        Resource,
+        ResourceCapabilities,
+    )
+
+    rid = _bench_ulid()
+    res = Resource(
+        resource_id=rid,
+        name=None,
+        resource_type_name=type_name,
+        classification=PhysVirt.PHYSICAL,
+        description="Bench resource",
+        properties={},
+        location=None,
+        content_hash=None,
+        sandbox_strategy=None,
+        capabilities=ResourceCapabilities(),
+        created_at=datetime.now(tz=UTC),
+        updated_at=datetime.now(tz=UTC),
+    )
+    res_repo.create(res)
+    return rid
+
+
+class TimeLinkChild:
+    """Benchmark ResourceRepository.link_child()."""
+
+    def setup(self) -> None:
+        from cleveragents.infrastructure.database.repositories import (
+            ResourceRepository,
+            ResourceTypeRepository,
+        )
+
+        self.factory = _setup_db()
+        self.rt_repo = ResourceTypeRepository(self.factory)
+        self.res_repo = ResourceRepository(self.factory)
+        _seed_types(self.rt_repo)
+        self.parent_id = _make_resource(self.res_repo, "bench/dag-parent")
+        self._child_ctr = 0
+
+    def time_link_child(self) -> None:
+        child_id = _make_resource(self.res_repo, "bench/dag-child")
+        self.res_repo.link_child(self.parent_id, child_id)
+
+
+class TimeUnlinkChild:
+    """Benchmark ResourceRepository.unlink_child()."""
+
+    def setup(self) -> None:
+        from cleveragents.infrastructure.database.repositories import (
+            ResourceRepository,
+            ResourceTypeRepository,
+        )
+
+        self.factory = _setup_db()
+        self.rt_repo = ResourceTypeRepository(self.factory)
+        self.res_repo = ResourceRepository(self.factory)
+        _seed_types(self.rt_repo)
+        self.parent_id = _make_resource(self.res_repo, "bench/dag-parent")
+        self.child_id = _make_resource(self.res_repo, "bench/dag-child")
+        self.res_repo.link_child(self.parent_id, self.child_id)
+
+    def time_unlink_child(self) -> None:
+        self.res_repo.unlink_child(self.parent_id, self.child_id)
+        # Re-link so the benchmark can run multiple times
+        self.res_repo.link_child(self.parent_id, self.child_id)
+
+
+class TimeAutoDiscoverChildren:
+    """Benchmark ResourceRepository.auto_discover_children()."""
+
+    def setup(self) -> None:
+        from cleveragents.infrastructure.database.repositories import (
+            ResourceRepository,
+            ResourceTypeRepository,
+        )
+
+        self.factory = _setup_db()
+        self.rt_repo = ResourceTypeRepository(self.factory)
+        self.res_repo = ResourceRepository(self.factory)
+        _seed_types(self.rt_repo)
+        self.parent_id = _make_resource(self.res_repo, "bench/dag-parent")
+
+    def time_auto_discover_children(self) -> None:
+        self.res_repo.auto_discover_children(self.parent_id)
+
+
+class TimeCycleDetection:
+    """Benchmark cycle detection with deep DAG chains."""
+
+    params: list[int] = [5, 10, 50]
+    param_names: list[str] = ["chain_depth"]
+
+    def setup(self, chain_depth: int) -> None:
+        from cleveragents.infrastructure.database.repositories import (
+            ResourceRepository,
+            ResourceTypeRepository,
+        )
+
+        self.factory = _setup_db()
+        self.rt_repo = ResourceTypeRepository(self.factory)
+        self.res_repo = ResourceRepository(self.factory)
+        _seed_types(
+            self.rt_repo,
+            "bench/chain-type",
+            "bench/chain-type",
+        )
+
+        # Build a chain: r0 -> r1 -> ... -> rN
+        self.chain_ids: list[str] = []
+        for _ in range(chain_depth + 1):
+            rid = _make_resource(self.res_repo, "bench/chain-type")
+            self.chain_ids.append(rid)
+
+        for i in range(chain_depth):
+            self.res_repo.link_child(
+                self.chain_ids[i],
+                self.chain_ids[i + 1],
+            )
+
+    def time_cycle_detection(self, chain_depth: int) -> None:
+        """Attempt to close the cycle (should raise)."""
+        try:
+            self.res_repo.link_child(
+                self.chain_ids[-1],
+                self.chain_ids[0],
+            )
+        except Exception:
+            pass

docs/reference/resource_dag.md

@@ -0,0 +1,157 @@
+# Resource DAG
+
+Resources in CleverAgents form a **directed acyclic graph** (DAG) where
+parent resources contain or reference child resources.  This document
+describes the linking rules, cycle detection, type compatibility
+enforcement, and auto-discovery behaviour.
+
+## DAG Rules
+
+| Rule | Description |
+|------|-------------|
+| **No self-loops** | A resource cannot be its own child. |
+| **No cycles** | If resource A is an ancestor of B, then B cannot become a parent of A. |
+| **Type compatibility** | The child's resource type must appear in the parent type's `child_types` list. |
+| **Unique links** | A given (parent, child) pair can only be linked once. |
+| **Both must exist** | Both the parent and the child resource must be registered before linking. |
+
+## API
+
+### `link_child(parent_id, child_id)`
+
+Links a child resource to a parent in the DAG.
+
+1. Validates that both resources exist in the registry.
+2. Checks **type compatibility** — the child resource's type must be
+   listed in the parent resource type's `child_types` field.
+3. Performs **cycle detection** — walks the ancestor chain of the
+   parent to ensure the child is not already an ancestor.
+4. Persists the link in the `resource_links` table.
+
+**Errors:**
+
+- `ResourceNotFoundRepoError` — parent or child does not exist.
+- `TypeIncompatibleError` — child type not in parent's `child_types`.
+- `CycleDetectedError` — linking would create a cycle.
+- `DuplicateResourceLinkError` — link already exists.
+
+### `unlink_child(parent_id, child_id)`
+
+Removes a parent-child link from the DAG.
+
+1. Validates that both resources exist.
+2. Validates the link exists.
+3. Deletes the link from `resource_links`.
+
+**Errors:**
+
+- `ResourceNotFoundRepoError` — parent or child does not exist.
+- `LinkNotFoundError` — the link does not exist.
+
+### `get_children(resource_id)`
+
+Returns all direct children of a resource (via `resource_links`).
+
+### `get_parents(resource_id)`
+
+Returns all direct parents of a resource (via `resource_links`).
+
+## Cycle Detection
+
+Cycle detection uses a **breadth-first search** upward through the
+`resource_links` table starting from the proposed parent.  If the
+proposed child is found among the ancestors, the link is rejected
+with a `CycleDetectedError` that includes the cycle path.
+
+### Example
+
+```
+A -> B -> C
+```
+
+Attempting to link `C -> A` would be rejected because `A` is an
+ancestor of `C`.  The error message includes the path:
+`A -> B -> C -> A`.
+
+## Type Compatibility
+
+Each resource type defines a `child_types` list of allowed child
+type names.  When linking, the system verifies:
+
+```
+child.resource_type_name in parent_type.child_types
+```
+
+If the parent type's `child_types` list is empty, **any** child type
+is allowed (no restriction).
+
+### Example
+
+```yaml
+# git-checkout type
+child_types: ["fs-directory", "git"]
+```
+
+Only resources of type `fs-directory` or `git` can be linked as
+children of a `git-checkout` resource.
+
+## Auto-Discovery
+
+### `auto_discover_children(resource_id)`
+
+Materializes child resources based on the parent's type auto-discovery
+configuration.
+
+1. Looks up the resource and its type.
+2. Reads the `auto_discovery` configuration from the type.
+3. For each discovery rule where `enabled` is `true`:
+   - Checks the child type exists in the database.
+   - Checks type compatibility with the parent.
+   - Creates a new child resource with `auto_discovered = true`.
+   - Links the child to the parent via `resource_links`.
+4. Returns the list of newly created child resources.
+
+### Auto-Discovery Configuration
+
+Auto-discovery is configured per resource type in YAML:
+
+```yaml
+auto_discovery:
+  enabled: true
+  rules:
+    - type: fs-directory
+      pattern: "*/"
+    - type: fs-file
+      pattern: "*"
+```
+
+Each rule specifies:
+
+- `type` — the child resource type name to create.
+- `pattern` — a glob pattern (used by handlers for actual file
+  discovery; the repository layer creates placeholder entries).
+
+### When Does Auto-Discovery Run?
+
+Auto-discovery is triggered by calling
+`auto_discover_children(resource_id)`.  This is typically invoked:
+
+- When a resource is first registered.
+- When a resource's contents change (e.g., new files appear).
+- On demand via CLI commands.
+
+## Database Schema
+
+### `resource_links` Table
+
+| Column | Type | Description |
+|--------|------|-------------|
+| `parent_id` | `String(26)` | FK to `resources.resource_id` |
+| `child_id` | `String(26)` | FK to `resources.resource_id` |
+| `created_at` | `String(30)` | ISO-8601 timestamp |
+
+Primary key: `(parent_id, child_id)`
+
+Constraints:
+- `parent_id != child_id` (no self-loops)
+- Foreign keys cascade on delete

features/resource_dag.feature

@@ -0,0 +1,166 @@
+@phase1 @domain @repository @resource_dag
+Feature: Resource DAG Linking and Discovery
+  As a system operator managing resource hierarchies
+  I want to link resources in a directed acyclic graph
+  So that parent-child relationships are enforced with type safety
+
+  Background:
+    Given a clean resource DAG database
+    And a resource type repository for DAG tests
+    And a resource repository for DAG tests
+
+  # ---------------------------------------------------------------------------
+  # link_child / unlink_child basic operations
+  # ---------------------------------------------------------------------------
+
+  @dag_link
+  Scenario: Link a child resource to a parent
+    Given a DAG parent type "dag/parent-type" allowing children '["dag/child-type"]'
+    And a DAG child type "dag/child-type"
+    And a DAG resource "P1" typed "dag/parent-type"
+    And a DAG resource "C1" typed "dag/child-type"
+    When DAG child "C1" is linked to parent "P1"
+    Then the DAG link should succeed without error
+    And DAG children of "P1" should include "C1"
+
+  @dag_unlink
+  Scenario: Unlink a child resource from a parent
+    Given a DAG parent type "dag/parent-type" allowing children '["dag/child-type"]'
+    And a DAG child type "dag/child-type"
+    And a DAG resource "P1" typed "dag/parent-type"
+    And a DAG resource "C1" typed "dag/child-type"
+    And DAG child "C1" is already linked to parent "P1"
+    When DAG child "C1" is unlinked from parent "P1"
+    Then the DAG unlink should succeed without error
+    And DAG children of "P1" should be empty
+
+  @dag_link @error_handling
+  Scenario: Linking a non-existent parent raises an error
+    Given a DAG child type "dag/child-type"
+    And a DAG resource "C1" typed "dag/child-type"
+    When DAG linking child "C1" to missing parent "MISSING_ID_00000000000000"
+    Then a DAG ResourceNotFoundRepoError should be raised
+
+  @dag_link @error_handling
+  Scenario: Linking a non-existent child raises an error
+    Given a DAG parent type "dag/parent-type" allowing children '["dag/child-type"]'
+    And a DAG resource "P1" typed "dag/parent-type"
+    When DAG linking missing child "MISSING_ID_00000000000000" to parent "P1"
+    Then a DAG ResourceNotFoundRepoError should be raised
+
+  @dag_link @error_handling
+  Scenario: Duplicate link raises an error
+    Given a DAG parent type "dag/parent-type" allowing children '["dag/child-type"]'
+    And a DAG child type "dag/child-type"
+    And a DAG resource "P1" typed "dag/parent-type"
+    And a DAG resource "C1" typed "dag/child-type"
+    And DAG child "C1" is already linked to parent "P1"
+    When DAG child "C1" is linked to parent "P1" again
+    Then a DAG DuplicateResourceLinkError should be raised
+
+  # ---------------------------------------------------------------------------
+  # Cycle detection
+  # ---------------------------------------------------------------------------
+
+  @dag_cycle
+  Scenario: Self-link is rejected
+    Given a DAG parent type "dag/parent-type" allowing children '["dag/parent-type"]'
+    And a DAG resource "P1" typed "dag/parent-type"
+    When DAG resource "P1" is linked to itself
+    Then a DAG CycleDetectedError should be raised
+
+  @dag_cycle
+  Scenario: Direct cycle A->B->A is rejected
+    Given a DAG parent type "dag/parent-type" allowing children '["dag/parent-type"]'
+    And a DAG resource "A" typed "dag/parent-type"
+    And a DAG resource "B" typed "dag/parent-type"
+    And DAG child "B" is already linked to parent "A"
+    When DAG child "A" is linked to parent "B"
+    Then a DAG CycleDetectedError should be raised
+
+  @dag_cycle
+  Scenario: Transitive cycle A->B->C->A is rejected
+    Given a DAG parent type "dag/parent-type" allowing children '["dag/parent-type"]'
+    And a DAG resource "A" typed "dag/parent-type"
+    And a DAG resource "B" typed "dag/parent-type"
+    And a DAG resource "C" typed "dag/parent-type"
+    And DAG child "B" is already linked to parent "A"
+    And DAG child "C" is already linked to parent "B"
+    When DAG child "A" is linked to parent "C"
+    Then a DAG CycleDetectedError should be raised
+
+  # ---------------------------------------------------------------------------
+  # Type compatibility enforcement
+  # ---------------------------------------------------------------------------
+
+  @dag_type_compat
+  Scenario: Linking incompatible types is rejected
+    Given a DAG parent type "dag/parent-type" allowing children '["dag/child-type"]'
+    And a DAG child type "dag/other-type"
+    And a DAG resource "P1" typed "dag/parent-type"
+    And a DAG resource "C1" typed "dag/other-type"
+    When DAG child "C1" is linked to parent "P1"
+    Then a DAG TypeIncompatibleError should be raised
+
+  @dag_type_compat
+  Scenario: Linking compatible types succeeds
+    Given a DAG parent type "dag/parent-type" allowing children '["dag/child-type"]'
+    And a DAG child type "dag/child-type"
+    And a DAG resource "P1" typed "dag/parent-type"
+    And a DAG resource "C1" typed "dag/child-type"
+    When DAG child "C1" is linked to parent "P1"
+    Then the DAG link should succeed without error
+
+  # ---------------------------------------------------------------------------
+  # auto_discover_children
+  # ---------------------------------------------------------------------------
+
+  @dag_auto_discover
+  Scenario: Auto-discover creates child resources per type rules
+    Given a DAG parent type "dag/discover-parent" with auto-discovery for "dag/discover-child"
+    And a DAG child type "dag/discover-child"
+    And a DAG resource "P1" typed "dag/discover-parent"
+    When DAG auto_discover_children is called for "P1"
+    Then at least 1 DAG child should be created
+    And DAG created children should be typed "dag/discover-child"
+    And DAG created children should be linked to "P1"
+
+  @dag_auto_discover
+  Scenario: Auto-discover with no rules creates nothing
+    Given a DAG parent type "dag/no-disc-parent" without auto-discovery
+    And a DAG resource "P1" typed "dag/no-disc-parent"
+    When DAG auto_discover_children is called for "P1"
+    Then 0 DAG children should be created
+
+  @dag_auto_discover @error_handling
+  Scenario: Auto-discover for non-existent resource raises error
+    When DAG auto_discover_children is called for missing "MISSING_ID_00000000000000"
+    Then a DAG ResourceNotFoundRepoError should be raised
+
+  # ---------------------------------------------------------------------------
+  # Tree traversal
+  # ---------------------------------------------------------------------------
+
+  @dag_traversal
+  Scenario: Get children returns all direct children
+    Given a DAG parent type "dag/parent-type" allowing children '["dag/child-type"]'
+    And a DAG child type "dag/child-type"
+    And a DAG resource "P1" typed "dag/parent-type"
+    And a DAG resource "C1" typed "dag/child-type"
+    And a DAG resource "C2" typed "dag/child-type"
+    And DAG child "C1" is already linked to parent "P1"
+    And DAG child "C2" is already linked to parent "P1"
+    When DAG children of "P1" are retrieved
+    Then 2 DAG children should be returned
+
+  @dag_traversal
+  Scenario: Get parents returns all direct parents
+    Given a DAG parent type "dag/parent-type" allowing children '["dag/child-type"]'
+    And a DAG child type "dag/child-type"
+    And a DAG resource "P1" typed "dag/parent-type"
+    And a DAG resource "P2" typed "dag/parent-type"
+    And a DAG resource "C1" typed "dag/child-type"
+    And DAG child "C1" is already linked to parent "P1"
+    And DAG child "C1" is already linked to parent "P2"
+    When DAG parents of "C1" are retrieved
+    Then 2 DAG parents should be returned

features/steps/resource_dag_steps.py

@@ -0,0 +1,441 @@
+"""Step definitions for resource_dag.feature."""
+
+from __future__ import annotations
+
+import json
+from datetime import UTC, datetime
+from typing import Any
+
+from behave import given, then, when
+from behave.runner import Context
+from sqlalchemy import create_engine, event
+from sqlalchemy.orm import sessionmaker
+
+from cleveragents.domain.models.core.resource import (
+    PhysVirt,
+    Resource,
+    ResourceCapabilities,
+)
+from cleveragents.domain.models.core.resource_type import (
+    ResourceKind,
+    ResourceTypeSpec,
+    SandboxStrategy,
+)
+from cleveragents.infrastructure.database.models import Base
+from cleveragents.infrastructure.database.repositories import (
+    CycleDetectedError,
+    DuplicateResourceLinkError,
+    LinkNotFoundError,
+    ResourceNotFoundRepoError,
+    ResourceRepository,
+    ResourceTypeRepository,
+    TypeIncompatibleError,
+)
+
+# Crockford base32 alphabet for generating ULIDs
+_CB32 = "0123456789ABCDEFGHJKMNPQRSTVWXYZ"
+_DAG_CTR = 0
+
+
+def _next_ulid() -> str:
+    """Return a unique, valid ULID string."""
+    global _DAG_CTR
+    _DAG_CTR += 1
+    n = _DAG_CTR
+    suffix = ""
+    for _ in range(16):
+        suffix = _CB32[n % 32] + suffix
+        n //= 32
+    return f"01HDAGT0AD{suffix}"[:26]
+
+
+def _make_type_spec(
+    name: str,
+    child_types: list[str] | None = None,
+    auto_discovery: dict[str, Any] | None = None,
+) -> ResourceTypeSpec:
+    """Create a ResourceTypeSpec."""
+    return ResourceTypeSpec(
+        name=name,
+        description=f"Test type {name}",
+        resource_kind=ResourceKind.PHYSICAL,
+        sandbox_strategy=SandboxStrategy.NONE,
+        user_addable=True,
+        cli_args=[],
+        parent_types=[],
+        child_types=child_types or [],
+        auto_discovery=auto_discovery,
+        equivalence=None,
+        handler=None,
+        capabilities={
+            "read": True,
+            "write": True,
+            "sandbox": True,
+            "checkpoint": False,
+        },
+        built_in=False,
+    )
+
+
+def _make_resource(
+    resource_id: str,
+    type_name: str,
+) -> Resource:
+    """Create a Resource domain object."""
+    return Resource(
+        resource_id=resource_id,
+        name=None,
+        resource_type_name=type_name,
+        classification=PhysVirt.PHYSICAL,
+        description=f"Test resource {resource_id}",
+        properties={},
+        location=None,
+        content_hash=None,
+        sandbox_strategy=None,
+        capabilities=ResourceCapabilities(),
+        created_at=datetime.now(tz=UTC),
+        updated_at=datetime.now(tz=UTC),
+    )
+
+
+# ── Background ─────────────────────────────────────────
+
+
+@given("a clean resource DAG database")
+def step_clean_dag_db(context: Context) -> None:
+    engine = create_engine("sqlite:///:memory:")
+
+    @event.listens_for(engine, "connect")
+    def _fk(conn: Any, _rec: Any) -> None:
+        conn.cursor().execute("PRAGMA foreign_keys=ON")
+
+    Base.metadata.create_all(engine)
+    context.dag_engine = engine  # type: ignore[attr-defined]
+    context.dag_factory = sessionmaker(  # type: ignore[attr-defined]
+        bind=engine
+    )
+    context.dag_resources = {}  # type: ignore[attr-defined]
+    context.dag_error = None  # type: ignore[attr-defined]
+    context.dag_result = None  # type: ignore[attr-defined]
+
+
+@given("a resource type repository for DAG tests")
+def step_rt_repo(context: Context) -> None:
+    context.dag_rt_repo = ResourceTypeRepository(  # type: ignore[attr-defined]
+        context.dag_factory  # type: ignore[attr-defined]
+    )
+
+
+@given("a resource repository for DAG tests")
+def step_res_repo(context: Context) -> None:
+    context.dag_res_repo = ResourceRepository(  # type: ignore[attr-defined]
+        context.dag_factory  # type: ignore[attr-defined]
+    )
+
+
+# ── Type setup ──────────────────────────────────────────
+
+
+@given("a DAG parent type \"{name}\" allowing children '{child_json}'")
+def step_dag_parent_type(context: Context, name: str, child_json: str) -> None:
+    child_list: list[str] = json.loads(child_json)
+    spec = _make_type_spec(name, child_types=child_list)
+    context.dag_rt_repo.create(spec)  # type: ignore[attr-defined]
+
+
+@given('a DAG child type "{name}"')
+def step_dag_child_type(context: Context, name: str) -> None:
+    spec = _make_type_spec(name)
+    context.dag_rt_repo.create(spec)  # type: ignore[attr-defined]
+
+
+@given('a DAG parent type "{name}" with auto-discovery for "{child_type}"')
+def step_dag_parent_auto(context: Context, name: str, child_type: str) -> None:
+    auto_disc = {
+        "enabled": True,
+        "rules": [{"type": child_type, "pattern": "*"}],
+    }
+    spec = _make_type_spec(
+        name,
+        child_types=[child_type],
+        auto_discovery=auto_disc,
+    )
+    context.dag_rt_repo.create(spec)  # type: ignore[attr-defined]
+
+
+@given('a DAG parent type "{name}" without auto-discovery')
+def step_dag_parent_no_disc(context: Context, name: str) -> None:
+    spec = _make_type_spec(name)
+    context.dag_rt_repo.create(spec)  # type: ignore[attr-defined]
+
+
+# ── Resource setup ──────────────────────────────────────
+
+
+@given('a DAG resource "{label}" typed "{type_name}"')
+def step_dag_resource(context: Context, label: str, type_name: str) -> None:
+    rid = _next_ulid()
+    res = _make_resource(rid, type_name)
+    context.dag_res_repo.create(res)  # type: ignore[attr-defined]
+    context.dag_resources[label] = rid  # type: ignore[attr-defined]
+
+
+# ── Link setup ──────────────────────────────────────────
+
+
+@given('DAG child "{child}" is already linked to parent "{parent}"')
+def step_dag_already_linked(context: Context, child: str, parent: str) -> None:
+    pid = context.dag_resources[parent]  # type: ignore[attr-defined]
+    cid = context.dag_resources[child]  # type: ignore[attr-defined]
+    context.dag_res_repo.link_child(pid, cid)  # type: ignore[attr-defined]
+
+
+# ── When: link / unlink ────────────────────────────────
+
+
+@when('DAG child "{child}" is linked to parent "{parent}"')
+def step_dag_link_child(context: Context, child: str, parent: str) -> None:
+    pid = context.dag_resources[parent]  # type: ignore[attr-defined]
+    cid = context.dag_resources[child]  # type: ignore[attr-defined]
+    try:
+        context.dag_res_repo.link_child(pid, cid)  # type: ignore[attr-defined]
+        context.dag_error = None  # type: ignore[attr-defined]
+    except (
+        CycleDetectedError,
+        TypeIncompatibleError,
+        DuplicateResourceLinkError,
+        ResourceNotFoundRepoError,
+    ) as exc:
+        context.dag_error = exc  # type: ignore[attr-defined]
+
+
+@when('DAG child "{child}" is linked to parent "{parent}" again')
+def step_dag_link_again(context: Context, child: str, parent: str) -> None:
+    pid = context.dag_resources[parent]  # type: ignore[attr-defined]
+    cid = context.dag_resources[child]  # type: ignore[attr-defined]
+    try:
+        context.dag_res_repo.link_child(pid, cid)  # type: ignore[attr-defined]
+        context.dag_error = None  # type: ignore[attr-defined]
+    except DuplicateResourceLinkError as exc:
+        context.dag_error = exc  # type: ignore[attr-defined]
+
+
+@when('DAG child "{child}" is unlinked from parent "{parent}"')
+def step_dag_unlink(context: Context, child: str, parent: str) -> None:
+    pid = context.dag_resources[parent]  # type: ignore[attr-defined]
+    cid = context.dag_resources[child]  # type: ignore[attr-defined]
+    try:
+        context.dag_res_repo.unlink_child(pid, cid)  # type: ignore[attr-defined]
+        context.dag_error = None  # type: ignore[attr-defined]
+    except (
+        ResourceNotFoundRepoError,
+        LinkNotFoundError,
+    ) as exc:
+        context.dag_error = exc  # type: ignore[attr-defined]
+
+
+@when('DAG linking child "{child}" to missing parent "{parent_id}"')
+def step_dag_link_missing_parent(context: Context, child: str, parent_id: str) -> None:
+    cid = context.dag_resources[child]  # type: ignore[attr-defined]
+    try:
+        context.dag_res_repo.link_child(parent_id, cid)  # type: ignore[attr-defined]
+        context.dag_error = None  # type: ignore[attr-defined]
+    except ResourceNotFoundRepoError as exc:
+        context.dag_error = exc  # type: ignore[attr-defined]
+
+
+@when('DAG linking missing child "{child_id}" to parent "{parent}"')
+def step_dag_link_missing_child(context: Context, child_id: str, parent: str) -> None:
+    pid = context.dag_resources[parent]  # type: ignore[attr-defined]
+    try:
+        context.dag_res_repo.link_child(pid, child_id)  # type: ignore[attr-defined]
+        context.dag_error = None  # type: ignore[attr-defined]
+    except ResourceNotFoundRepoError as exc:
+        context.dag_error = exc  # type: ignore[attr-defined]
+
+
+@when('DAG resource "{res}" is linked to itself')
+def step_dag_self_link(context: Context, res: str) -> None:
+    rid = context.dag_resources[res]  # type: ignore[attr-defined]
+    try:
+        context.dag_res_repo.link_child(rid, rid)  # type: ignore[attr-defined]
+        context.dag_error = None  # type: ignore[attr-defined]
+    except CycleDetectedError as exc:
+        context.dag_error = exc  # type: ignore[attr-defined]
+
+
+# ── When: auto_discover ─────────────────────────────────
+
+
+@when('DAG auto_discover_children is called for "{label}"')
+def step_dag_auto_discover(context: Context, label: str) -> None:
+    rid = context.dag_resources[label]  # type: ignore[attr-defined]
+    try:
+        context.dag_result = (  # type: ignore[attr-defined]
+            context.dag_res_repo.auto_discover_children(  # type: ignore[attr-defined]
+                rid
+            )
+        )
+        context.dag_error = None  # type: ignore[attr-defined]
+    except ResourceNotFoundRepoError as exc:
+        context.dag_error = exc  # type: ignore[attr-defined]
+        context.dag_result = []  # type: ignore[attr-defined]
+
+
+@when('DAG auto_discover_children is called for missing "{resource_id}"')
+def step_dag_auto_discover_missing(context: Context, resource_id: str) -> None:
+    try:
+        context.dag_result = (  # type: ignore[attr-defined]
+            context.dag_res_repo.auto_discover_children(  # type: ignore[attr-defined]
+                resource_id
+            )
+        )
+        context.dag_error = None  # type: ignore[attr-defined]
+    except ResourceNotFoundRepoError as exc:
+        context.dag_error = exc  # type: ignore[attr-defined]
+        context.dag_result = []  # type: ignore[attr-defined]
+
+
+# ── When: traversal ─────────────────────────────────────
+
+
+@when('DAG children of "{label}" are retrieved')
+def step_dag_get_children(context: Context, label: str) -> None:
+    rid = context.dag_resources[label]  # type: ignore[attr-defined]
+    context.dag_result = (  # type: ignore[attr-defined]
+        context.dag_res_repo.get_children(rid)  # type: ignore[attr-defined]
+    )
+
+
+@when('DAG parents of "{label}" are retrieved')
+def step_dag_get_parents(context: Context, label: str) -> None:
+    rid = context.dag_resources[label]  # type: ignore[attr-defined]
+    context.dag_result = (  # type: ignore[attr-defined]
+        context.dag_res_repo.get_parents(rid)  # type: ignore[attr-defined]
+    )
+
+
+# ── Then: assertions ────────────────────────────────────
+
+
+@then("the DAG link should succeed without error")
+def step_dag_link_ok(context: Context) -> None:
+    assert context.dag_error is None, (  # type: ignore[attr-defined]
+        f"Expected no error, got: {context.dag_error}"  # type: ignore[attr-defined]
+    )
+
+
+@then("the DAG unlink should succeed without error")
+def step_dag_unlink_ok(context: Context) -> None:
+    assert context.dag_error is None, (  # type: ignore[attr-defined]
+        f"Expected no error, got: {context.dag_error}"  # type: ignore[attr-defined]
+    )
+
+
+@then('DAG children of "{label}" should include "{child}"')
+def step_dag_children_include(context: Context, label: str, child: str) -> None:
+    pid = context.dag_resources[label]  # type: ignore[attr-defined]
+    cid = context.dag_resources[child]  # type: ignore[attr-defined]
+    children = context.dag_res_repo.get_children(pid)  # type: ignore[attr-defined]
+    child_ids = [c.resource_id for c in children]
+    assert cid in child_ids, f"Child {cid} not in children {child_ids}"
+
+
+@then('DAG children of "{label}" should be empty')
+def step_dag_children_empty(context: Context, label: str) -> None:
+    pid = context.dag_resources[label]  # type: ignore[attr-defined]
+    children = context.dag_res_repo.get_children(pid)  # type: ignore[attr-defined]
+    assert len(children) == 0, f"Expected 0 children, got {len(children)}"
+
+
+@then("a DAG ResourceNotFoundRepoError should be raised")
+def step_dag_not_found(context: Context) -> None:
+    assert isinstance(
+        context.dag_error,  # type: ignore[attr-defined]
+        ResourceNotFoundRepoError,
+    ), (
+        f"Expected ResourceNotFoundRepoError, got "
+        f"{type(context.dag_error).__name__}: "  # type: ignore[attr-defined]
+        f"{context.dag_error}"  # type: ignore[attr-defined]
+    )
+
+
+@then("a DAG CycleDetectedError should be raised")
+def step_dag_cycle_error(context: Context) -> None:
+    assert isinstance(
+        context.dag_error,  # type: ignore[attr-defined]
+        CycleDetectedError,
+    ), (
+        f"Expected CycleDetectedError, got "
+        f"{type(context.dag_error).__name__}: "  # type: ignore[attr-defined]
+        f"{context.dag_error}"  # type: ignore[attr-defined]
+    )
+
+
+@then("a DAG TypeIncompatibleError should be raised")
+def step_dag_type_error(context: Context) -> None:
+    assert isinstance(
+        context.dag_error,  # type: ignore[attr-defined]
+        TypeIncompatibleError,
+    ), (
+        f"Expected TypeIncompatibleError, got "
+        f"{type(context.dag_error).__name__}: "  # type: ignore[attr-defined]
+        f"{context.dag_error}"  # type: ignore[attr-defined]
+    )
+
+
+@then("a DAG DuplicateResourceLinkError should be raised")
+def step_dag_dup_link(context: Context) -> None:
+    assert isinstance(
+        context.dag_error,  # type: ignore[attr-defined]
+        DuplicateResourceLinkError,
+    ), (
+        f"Expected DuplicateResourceLinkError, got "
+        f"{type(context.dag_error).__name__}: "  # type: ignore[attr-defined]
+        f"{context.dag_error}"  # type: ignore[attr-defined]
+    )
+
+
+@then("at least {count:d} DAG child should be created")
+def step_dag_at_least_n(context: Context, count: int) -> None:
+    result = context.dag_result or []  # type: ignore[attr-defined]
+    assert len(result) >= count, f"Expected >= {count} children, got {len(result)}"
+
+
+@then("{count:d} DAG children should be created")
+def step_dag_exact_n(context: Context, count: int) -> None:
+    result = context.dag_result or []  # type: ignore[attr-defined]
+    assert len(result) == count, f"Expected {count} children, got {len(result)}"
+
+
+@then('DAG created children should be typed "{type_name}"')
+def step_dag_children_typed(context: Context, type_name: str) -> None:
+    result = context.dag_result or []  # type: ignore[attr-defined]
+    for child in result:
+        assert child.resource_type_name == type_name, (
+            f"Expected type {type_name}, got {child.resource_type_name}"
+        )
+
+
+@then('DAG created children should be linked to "{label}"')
+def step_dag_children_linked(context: Context, label: str) -> None:
+    pid = context.dag_resources[label]  # type: ignore[attr-defined]
+    children = context.dag_res_repo.get_children(pid)  # type: ignore[attr-defined]
+    result = context.dag_result or []  # type: ignore[attr-defined]
+    child_ids = {c.resource_id for c in children}
+    for created in result:
+        assert created.resource_id in child_ids, (
+            f"Created child {created.resource_id} not linked to parent {pid}"
+        )
+
+
+@then("{count:d} DAG children should be returned")
+def step_dag_n_children(context: Context, count: int) -> None:
+    result = context.dag_result or []  # type: ignore[attr-defined]
+    assert len(result) == count, f"Expected {count} children, got {len(result)}"
+
+
+@then("{count:d} DAG parents should be returned")
+def step_dag_n_parents(context: Context, count: int) -> None:
+    result = context.dag_result or []  # type: ignore[attr-defined]
+    assert len(result) == count, f"Expected {count} parents, got {len(result)}"

implementation_plan.md

@@ -2407,20 +2407,20 @@ No standalone Q0-Advanced commits planned. Advanced QA enhancements are bundled
   - [ ] Git [Jeff]: `git branch -d feature/m2-resource-types`
   - [x] 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%.
 
-- [ ] **COMMIT (Owner: Jeff | Group: B1.dag | Branch: feature/m2-resource-dag | Planned: Day 12 | Expected: Day 16) - Commit message: "feat(resource): add DAG linking and discovery"**
-  - [ ] Git [Jeff]: `git checkout master`
-  - [ ] Git [Jeff]: `git pull origin master`
-  - [ ] Git [Jeff]: `git checkout -b feature/m2-resource-dag`
-  - [ ] Git [Jeff]: `git fetch origin && git merge origin/master` (run before final tests and before commit)
-  - [ ] Code [Jeff]: Implement `link_child`/`unlink_child` in `ResourceRepository` with cycle detection and type compatibility enforcement.
-  - [ ] Code [Jeff]: Add `auto_discover_children(resource_id)` that materializes child resources per type rules.
-  - [ ] Docs [Jeff]: Document DAG rules and auto-discovery behavior in `docs/reference/resource_dag.md`.
-  - [ ] Tests (Behave) [Jeff]: Add scenarios for link/unlink, cycle rejection, and auto_discover creation.
-  - [ ] Tests (Robot) [Jeff]: Add Robot test that links a child and verifies tree output ordering.
-  - [ ] Tests (ASV) [Jeff]: Add `benchmarks/resource_dag_bench.py` for link/auto_discover performance.
-  - [ ] Quality [Jeff]: Run `nox` (all default sessions, including benchmark), fix any errors if needed ensuring nox passes.
-  - [ ] Git [Jeff]: `git add .`
-  - [ ] Git [Jeff]: `git commit -m "feat(resource): add DAG linking and discovery"`
+- [x] **COMMIT (Owner: Jeff | Group: B1.dag | Branch: feature/m2-resource-dag | Planned: Day 12 | Expected: Day 16) - Commit message: "feat(resource): add DAG linking and discovery"**
+  - [x] Git [Jeff]: `git checkout master`
+  - [x] Git [Jeff]: `git pull origin master`
+  - [x] Git [Jeff]: `git checkout -b feature/m2-resource-dag`
+  - [x] Git [Jeff]: `git fetch origin && git merge origin/master` (run before final tests and before commit)
+  - [x] Code [Jeff]: Implement `link_child`/`unlink_child` in `ResourceRepository` with cycle detection and type compatibility enforcement.
+  - [x] Code [Jeff]: Add `auto_discover_children(resource_id)` that materializes child resources per type rules.
+  - [x] Docs [Jeff]: Document DAG rules and auto-discovery behavior in `docs/reference/resource_dag.md`.
+  - [x] Tests (Behave) [Jeff]: Add scenarios for link/unlink, cycle rejection, and auto_discover creation.
+  - [x] Tests (Robot) [Jeff]: Add Robot test that links a child and verifies tree output ordering.
+  - [x] Tests (ASV) [Jeff]: Add `benchmarks/resource_dag_bench.py` for link/auto_discover performance.
+  - [x] Quality [Jeff]: Run `nox` (all default sessions, including benchmark), fix any errors if needed ensuring nox passes.
+  - [x] Git [Jeff]: `git add .`
+  - [x] Git [Jeff]: `git commit -m "feat(resource): add DAG linking and discovery"`
   - [ ] Forgejo PR [Jeff]: Open PR from `feature/m2-resource-dag` to `master` with description "Add resource DAG linking, cycle checks, and auto-discovery with tests/docs.".
   - [ ] Git [Jeff]: `git checkout master`
   - [ ] Git [Jeff]: `git branch -d feature/m2-resource-dag`

robot/resource_dag.robot

@@ -0,0 +1,112 @@
+*** Settings ***
+Documentation       Resource DAG Linking and Discovery Tests
+Library             Process
+Library             OperatingSystem
+
+*** Variables ***
+${PYTHON}           python
+
+*** Test Cases ***
+Link Child And Verify Tree
+    [Documentation]    Link a child resource and verify it appears in children list
+    ${script}=    Catenate    SEPARATOR=\n
+    ...    import json
+    ...    from datetime import datetime, UTC
+    ...    from sqlalchemy import create_engine, event
+    ...    from sqlalchemy.orm import sessionmaker
+    ...    from cleveragents.infrastructure.database.models import Base
+    ...    from cleveragents.infrastructure.database.repositories import ResourceTypeRepository, ResourceRepository
+    ...    from cleveragents.domain.models.core.resource_type import ResourceTypeSpec, ResourceKind, SandboxStrategy
+    ...    from cleveragents.domain.models.core.resource import Resource, PhysVirt, ResourceCapabilities
+    ...    engine = create_engine("sqlite:///:memory:")
+    ...    @event.listens_for(engine, "connect")
+    ...    def _fk(conn, _): conn.cursor().execute("PRAGMA foreign_keys=ON")
+    ...    Base.metadata.create_all(engine)
+    ...    factory = sessionmaker(bind=engine)
+    ...    rt_repo = ResourceTypeRepository(factory)
+    ...    res_repo = ResourceRepository(factory)
+    ...    parent_spec = ResourceTypeSpec(name="robot/dag-parent", description="Parent", resource_kind=ResourceKind.PHYSICAL, sandbox_strategy=SandboxStrategy.NONE, user_addable=True, cli_args=[], parent_types=[], child_types=["robot/dag-child"], auto_discovery=None, equivalence=None, handler=None, capabilities={"read": True, "write": True, "sandbox": True, "checkpoint": False}, built_in=False)
+    ...    child_spec = ResourceTypeSpec(name="robot/dag-child", description="Child", resource_kind=ResourceKind.PHYSICAL, sandbox_strategy=SandboxStrategy.NONE, user_addable=True, cli_args=[], parent_types=[], child_types=[], auto_discovery=None, equivalence=None, handler=None, capabilities={"read": True, "write": True, "sandbox": True, "checkpoint": False}, built_in=False)
+    ...    rt_repo.create(parent_spec)
+    ...    rt_repo.create(child_spec)
+    ...    p = Resource(resource_id="01HDAGR0B0T0000000PARENT01", name=None, resource_type_name="robot/dag-parent", classification=PhysVirt.PHYSICAL, properties={}, location=None, capabilities=ResourceCapabilities(), created_at=datetime.now(tz=UTC), updated_at=datetime.now(tz=UTC))
+    ...    c = Resource(resource_id="01HDAGR0B0T000000CHILD001", name=None, resource_type_name="robot/dag-child", classification=PhysVirt.PHYSICAL, properties={}, location=None, capabilities=ResourceCapabilities(), created_at=datetime.now(tz=UTC), updated_at=datetime.now(tz=UTC))
+    ...    res_repo.create(p)
+    ...    res_repo.create(c)
+    ...    res_repo.link_child("01HDAGR0B0T0000000PARENT01", "01HDAGR0B0T000000CHILD001")
+    ...    children = res_repo.get_children("01HDAGR0B0T0000000PARENT01")
+    ...    assert len(children) == 1, f"Expected 1 child, got {len(children)}"
+    ...    assert children[0].resource_id == "01HDAGR0B0T000000CHILD001"
+    ...    print("Link child and verify tree passed")
+    ${result}=    Run Process    ${PYTHON}    -c    ${script}
+    Should Be Equal As Integers    ${result.rc}    0    Link test failed: ${result.stderr}
+    Should Contain    ${result.stdout}    Link child and verify tree passed
+
+Cycle Detection Rejects A To B To A
+    [Documentation]    Linking A->B then B->A should raise CycleDetectedError
+    ${script}=    Catenate    SEPARATOR=\n
+    ...    from datetime import datetime, UTC
+    ...    from sqlalchemy import create_engine, event
+    ...    from sqlalchemy.orm import sessionmaker
+    ...    from cleveragents.infrastructure.database.models import Base
+    ...    from cleveragents.infrastructure.database.repositories import ResourceTypeRepository, ResourceRepository, CycleDetectedError
+    ...    from cleveragents.domain.models.core.resource_type import ResourceTypeSpec, ResourceKind, SandboxStrategy
+    ...    from cleveragents.domain.models.core.resource import Resource, PhysVirt, ResourceCapabilities
+    ...    engine = create_engine("sqlite:///:memory:")
+    ...    @event.listens_for(engine, "connect")
+    ...    def _fk(conn, _): conn.cursor().execute("PRAGMA foreign_keys=ON")
+    ...    Base.metadata.create_all(engine)
+    ...    factory = sessionmaker(bind=engine)
+    ...    rt_repo = ResourceTypeRepository(factory)
+    ...    res_repo = ResourceRepository(factory)
+    ...    spec = ResourceTypeSpec(name="robot/cycle-type", description="Cycle", resource_kind=ResourceKind.PHYSICAL, sandbox_strategy=SandboxStrategy.NONE, user_addable=True, cli_args=[], parent_types=[], child_types=["robot/cycle-type"], auto_discovery=None, equivalence=None, handler=None, capabilities={"read": True, "write": True, "sandbox": True, "checkpoint": False}, built_in=False)
+    ...    rt_repo.create(spec)
+    ...    a = Resource(resource_id="01HDAGR0B0TCYCLE0000000A1", name=None, resource_type_name="robot/cycle-type", classification=PhysVirt.PHYSICAL, properties={}, location=None, capabilities=ResourceCapabilities(), created_at=datetime.now(tz=UTC), updated_at=datetime.now(tz=UTC))
+    ...    b = Resource(resource_id="01HDAGR0B0TCYCLE0000000B1", name=None, resource_type_name="robot/cycle-type", classification=PhysVirt.PHYSICAL, properties={}, location=None, capabilities=ResourceCapabilities(), created_at=datetime.now(tz=UTC), updated_at=datetime.now(tz=UTC))
+    ...    res_repo.create(a)
+    ...    res_repo.create(b)
+    ...    res_repo.link_child("01HDAGR0B0TCYCLE0000000A1", "01HDAGR0B0TCYCLE0000000B1")
+    ...    try:
+    ...        res_repo.link_child("01HDAGR0B0TCYCLE0000000B1", "01HDAGR0B0TCYCLE0000000A1")
+    ...        assert False, "Should have raised CycleDetectedError"
+    ...    except CycleDetectedError:
+    ...        print("Cycle detection passed")
+    ${result}=    Run Process    ${PYTHON}    -c    ${script}
+    Should Be Equal As Integers    ${result.rc}    0    Cycle test failed: ${result.stderr}
+    Should Contain    ${result.stdout}    Cycle detection passed
+
+Auto Discover Children
+    [Documentation]    Auto-discover creates child resources per type rules
+    ${script}=    Catenate    SEPARATOR=\n
+    ...    import json
+    ...    from datetime import datetime, UTC
+    ...    from sqlalchemy import create_engine, event
+    ...    from sqlalchemy.orm import sessionmaker
+    ...    from cleveragents.infrastructure.database.models import Base
+    ...    from cleveragents.infrastructure.database.repositories import ResourceTypeRepository, ResourceRepository
+    ...    from cleveragents.domain.models.core.resource_type import ResourceTypeSpec, ResourceKind, SandboxStrategy
+    ...    from cleveragents.domain.models.core.resource import Resource, PhysVirt, ResourceCapabilities
+    ...    engine = create_engine("sqlite:///:memory:")
+    ...    @event.listens_for(engine, "connect")
+    ...    def _fk(conn, _): conn.cursor().execute("PRAGMA foreign_keys=ON")
+    ...    Base.metadata.create_all(engine)
+    ...    factory = sessionmaker(bind=engine)
+    ...    rt_repo = ResourceTypeRepository(factory)
+    ...    res_repo = ResourceRepository(factory)
+    ...    parent_spec = ResourceTypeSpec(name="robot/disc-parent", description="Discoverer", resource_kind=ResourceKind.PHYSICAL, sandbox_strategy=SandboxStrategy.NONE, user_addable=True, cli_args=[], parent_types=[], child_types=["robot/disc-child"], auto_discovery={"enabled": True, "rules": [{"type": "robot/disc-child", "pattern": "*"}]}, equivalence=None, handler=None, capabilities={"read": True, "write": True, "sandbox": True, "checkpoint": False}, built_in=False)
+    ...    child_spec = ResourceTypeSpec(name="robot/disc-child", description="Discovered", resource_kind=ResourceKind.PHYSICAL, sandbox_strategy=SandboxStrategy.NONE, user_addable=True, cli_args=[], parent_types=[], child_types=[], auto_discovery=None, equivalence=None, handler=None, capabilities={"read": True, "write": True, "sandbox": True, "checkpoint": False}, built_in=False)
+    ...    rt_repo.create(parent_spec)
+    ...    rt_repo.create(child_spec)
+    ...    p = Resource(resource_id="01HDAGR0B0TDISC00PARENT01", name=None, resource_type_name="robot/disc-parent", classification=PhysVirt.PHYSICAL, properties={}, location=None, capabilities=ResourceCapabilities(), created_at=datetime.now(tz=UTC), updated_at=datetime.now(tz=UTC))
+    ...    res_repo.create(p)
+    ...    created = res_repo.auto_discover_children("01HDAGR0B0TDISC00PARENT01")
+    ...    assert len(created) >= 1, f"Expected >=1 children, got {len(created)}"
+    ...    assert created[0].resource_type_name == "robot/disc-child"
+    ...    children = res_repo.get_children("01HDAGR0B0TDISC00PARENT01")
+    ...    assert len(children) >= 1, f"Expected >=1 linked children, got {len(children)}"
+    ...    print("Auto discover children passed")
+    ${result}=    Run Process    ${PYTHON}    -c    ${script}
+    Should Be Equal As Integers    ${result.rc}    0    Auto discover test failed: ${result.stderr}
+    Should Contain    ${result.stdout}    Auto discover children passed
+
+*** Keywords ***

src/cleveragents/infrastructure/database/models.py

@@ -17,6 +17,7 @@ Alembic migrations.
 | ``resource_types`` | ``ResourceTypeModel`` | Resource type defs |
 | ``resources`` | ``ResourceModel`` | Resource instances |
 | ``resource_edges`` | ``ResourceEdgeModel`` | Resource DAG edges |
+| ``resource_links`` | ``ResourceLinkModel`` | Validated links |
 | ``ns_projects`` | ``NamespacedProjectModel`` | Namespaced projects |
 | ``project_resource_links`` | ``ProjectResourceLinkModel`` | Project-resource links |
 | ``tools`` | ``ToolModel`` | Tool registry entries |
@@ -1466,6 +1467,51 @@ class ResourceEdgeModel(Base):  # type: ignore[misc]
     )
 
 
+# ---------------------------------------------------------------------------
+# Resource Link Models (Stage B1 - migration b1_001_resource_links)
+# ---------------------------------------------------------------------------
+
+
+class ResourceLinkModel(Base):  # type: ignore[misc]
+    """Database model for validated resource DAG links.
+
+    Stores parent-child links between resources after validation
+    (cycle detection, type compatibility).  Unlike ``resource_edges``
+    which stores raw DAG edges with link-type metadata, this table
+    records validated DAG relationships managed by ``link_child`` /
+    ``unlink_child``.
+
+    Table: ``resource_links``
+    """
+
+    __allow_unmapped__ = True
+    __tablename__ = "resource_links"
+
+    # 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,
+    )
+
+    # Timestamp (ISO-8601 string)
+    created_at = Column(String(30), nullable=False)
+
+    __table_args__ = (
+        CheckConstraint(
+            "parent_id != child_id",
+            name="ck_resource_links_no_self_loop",
+        ),
+        Index("ix_resource_links_child", "child_id"),
+        Index("ix_resource_links_parent", "parent_id"),
+    )
+
+
 # ---------------------------------------------------------------------------
 # Tool Registry Models (Stage C1 - migration c1_001_tool_registry)
 # ---------------------------------------------------------------------------

src/cleveragents/infrastructure/database/repositories.py

@@ -90,6 +90,7 @@ from cleveragents.infrastructure.database.models import (
     ProjectModel,
     ProjectResourceLinkModel,
     ResourceEdgeModel,
+    ResourceLinkModel,
     ResourceModel,
     ResourceTypeModel,
     ToolBindingModel,
@@ -1560,6 +1561,52 @@ class DuplicateResourceError(DatabaseError):
         self.resource_name = name
 
 
+class CycleDetectedError(BusinessRuleViolation):
+    """Raised when linking would create a cycle in the resource DAG."""
+
+    def __init__(self, parent_id: str, child_id: str, path: list[str]):
+        cycle_str = " -> ".join(path)
+        super().__init__(
+            f"Linking {parent_id} -> {child_id} would create a cycle: {cycle_str}"
+        )
+        self.parent_id = parent_id
+        self.child_id = child_id
+        self.path = path
+
+
+class TypeIncompatibleError(BusinessRuleViolation):
+    """Raised when child type is not in parent type's child_types."""
+
+    def __init__(
+        self,
+        parent_type: str,
+        child_type: str,
+    ):
+        super().__init__(
+            f"Type '{child_type}' is not an allowed child of '{parent_type}'"
+        )
+        self.parent_type = parent_type
+        self.child_type = child_type
+
+
+class LinkNotFoundError(DatabaseError):
+    """Raised when a resource link does not exist."""
+
+    def __init__(self, parent_id: str, child_id: str):
+        super().__init__(f"Link from '{parent_id}' to '{child_id}' not found")
+        self.parent_id = parent_id
+        self.child_id = child_id
+
+
+class DuplicateResourceLinkError(DatabaseError):
+    """Raised when a resource link already exists."""
+
+    def __init__(self, parent_id: str, child_id: str):
+        super().__init__(f"Link from '{parent_id}' to '{child_id}' already exists")
+        self.parent_id = parent_id
+        self.child_id = child_id
+
+
 class ResourceTypeRepository:
     """Repository for resource type persistence.
 
@@ -2216,6 +2263,430 @@ class ResourceRepository:
                 f"Failed to resolve resource '{name_or_id}': {exc}"
             ) from exc
 
+    @database_retry
+    def link_child(self, parent_id: str, child_id: str) -> None:
+        """Link a child resource to a parent in the DAG.
+
+        Validates both resources exist, checks type compatibility
+        (child's type must be in parent type's ``child_types``),
+        and detects cycles before persisting.
+
+        Args:
+            parent_id: ULID of the parent resource.
+            child_id: ULID of the child resource.
+
+        Raises:
+            ResourceNotFoundRepoError: If either resource is missing.
+            TypeIncompatibleError: If child type is not allowed.
+            CycleDetectedError: If the link would create a cycle.
+            DuplicateResourceLinkError: If the link already exists.
+            DatabaseError: On transient or unexpected DB errors.
+        """
+        if parent_id == child_id:
+            raise CycleDetectedError(parent_id, child_id, [parent_id, child_id])
+
+        session = self._session()
+        try:
+            parent_row = (
+                session.query(ResourceModel).filter_by(resource_id=parent_id).first()
+            )
+            if parent_row is None:
+                raise ResourceNotFoundRepoError(parent_id)
+
+            child_row = (
+                session.query(ResourceModel).filter_by(resource_id=child_id).first()
+            )
+            if child_row is None:
+                raise ResourceNotFoundRepoError(child_id)
+
+            # Check type compatibility
+            parent_type_name = cast(str, parent_row.type_name)
+            child_type_name = cast(str, child_row.type_name)
+
+            parent_type_row = (
+                session.query(ResourceTypeModel)
+                .filter_by(name=parent_type_name)
+                .first()
+            )
+            if parent_type_row is not None:
+                allowed_raw = cast(
+                    "str | None",
+                    parent_type_row.allowed_child_types_json,
+                )
+                allowed_children: list[str] = (
+                    json.loads(allowed_raw) if allowed_raw else []
+                )
+                if allowed_children and child_type_name not in allowed_children:
+                    raise TypeIncompatibleError(parent_type_name, child_type_name)
+
+            # Check for duplicate link
+            existing = (
+                session.query(ResourceLinkModel)
+                .filter_by(parent_id=parent_id, child_id=child_id)
+                .first()
+            )
+            if existing is not None:
+                raise DuplicateResourceLinkError(parent_id, child_id)
+
+            # Cycle detection: ensure child_id is not an
+            # ancestor of parent_id
+            ancestors = self._get_ancestors(session, parent_id)
+            if child_id in ancestors:
+                cycle_path = self._build_cycle_path(session, parent_id, child_id)
+                raise CycleDetectedError(parent_id, child_id, cycle_path)
+
+            link = ResourceLinkModel(
+                parent_id=parent_id,
+                child_id=child_id,
+                created_at=datetime.now(tz=UTC).isoformat(),
+            )
+            session.add(link)
+            session.flush()
+        except (
+            ResourceNotFoundRepoError,
+            TypeIncompatibleError,
+            CycleDetectedError,
+            DuplicateResourceLinkError,
+        ):
+            raise
+        except IntegrityError as exc:
+            session.rollback()
+            raise DatabaseError(
+                f"Failed to link {parent_id} -> {child_id}: {exc}"
+            ) from exc
+        except (
+            OperationalError,
+            SQLAlchemyDatabaseError,
+        ) as exc:
+            session.rollback()
+            raise DatabaseError(
+                f"Failed to link {parent_id} -> {child_id}: {exc}"
+            ) from exc
+
+    @database_retry
+    def unlink_child(self, parent_id: str, child_id: str) -> None:
+        """Remove a parent-child link from the DAG.
+
+        Args:
+            parent_id: ULID of the parent resource.
+            child_id: ULID of the child resource.
+
+        Raises:
+            ResourceNotFoundRepoError: If either resource missing.
+            LinkNotFoundError: If the link does not exist.
+            DatabaseError: On transient or unexpected DB errors.
+        """
+        session = self._session()
+        try:
+            parent_row = (
+                session.query(ResourceModel).filter_by(resource_id=parent_id).first()
+            )
+            if parent_row is None:
+                raise ResourceNotFoundRepoError(parent_id)
+
+            child_row = (
+                session.query(ResourceModel).filter_by(resource_id=child_id).first()
+            )
+            if child_row is None:
+                raise ResourceNotFoundRepoError(child_id)
+
+            link = (
+                session.query(ResourceLinkModel)
+                .filter_by(parent_id=parent_id, child_id=child_id)
+                .first()
+            )
+            if link is None:
+                raise LinkNotFoundError(parent_id, child_id)
+
+            session.delete(link)
+            session.flush()
+        except (
+            ResourceNotFoundRepoError,
+            LinkNotFoundError,
+        ):
+            raise
+        except (
+            OperationalError,
+            SQLAlchemyDatabaseError,
+        ) as exc:
+            session.rollback()
+            raise DatabaseError(
+                f"Failed to unlink {parent_id} -> {child_id}: {exc}"
+            ) from exc
+
+    @database_retry
+    def get_children(self, resource_id: str) -> list[Any]:
+        """Get all direct children of a resource.
+
+        Args:
+            resource_id: ULID of the parent resource.
+
+        Returns:
+            List of child ``Resource`` domain objects.
+        """
+        session = self._session()
+        try:
+            links = (
+                session.query(ResourceLinkModel).filter_by(parent_id=resource_id).all()
+            )
+            children: list[Any] = []
+            for link in links:
+                child_row = (
+                    session.query(ResourceModel)
+                    .filter_by(resource_id=cast(str, link.child_id))
+                    .first()
+                )
+                if child_row is not None:
+                    children.append(self._to_domain(child_row))
+            return children
+        except (
+            OperationalError,
+            SQLAlchemyDatabaseError,
+        ) as exc:
+            raise DatabaseError(
+                f"Failed to get children of '{resource_id}': {exc}"
+            ) from exc
+
+    @database_retry
+    def get_parents(self, resource_id: str) -> list[Any]:
+        """Get all direct parents of a resource.
+
+        Args:
+            resource_id: ULID of the child resource.
+
+        Returns:
+            List of parent ``Resource`` domain objects.
+        """
+        session = self._session()
+        try:
+            links = (
+                session.query(ResourceLinkModel).filter_by(child_id=resource_id).all()
+            )
+            parents: list[Any] = []
+            for link in links:
+                parent_row = (
+                    session.query(ResourceModel)
+                    .filter_by(resource_id=cast(str, link.parent_id))
+                    .first()
+                )
+                if parent_row is not None:
+                    parents.append(self._to_domain(parent_row))
+            return parents
+        except (
+            OperationalError,
+            SQLAlchemyDatabaseError,
+        ) as exc:
+            raise DatabaseError(
+                f"Failed to get parents of '{resource_id}': {exc}"
+            ) from exc
+
+    @database_retry
+    def auto_discover_children(self, resource_id: str) -> list[Any]:
+        """Materialize child resources per type auto-discovery.
+
+        Looks up the resource's type, checks auto_discovery config,
+        and for each child type with auto-discover enabled, creates
+        a child resource and links it to the parent.
+
+        Args:
+            resource_id: ULID of the parent resource.
+
+        Returns:
+            List of newly created child ``Resource`` domain objects.
+
+        Raises:
+            ResourceNotFoundRepoError: If the resource is missing.
+            DatabaseError: On transient or unexpected DB errors.
+        """
+        from ulid import ULID as _ULID
+
+        from cleveragents.domain.models.core.resource import (
+            PhysVirt,
+            Resource,
+            ResourceCapabilities,
+        )
+
+        session = self._session()
+        try:
+            parent_row = (
+                session.query(ResourceModel).filter_by(resource_id=resource_id).first()
+            )
+            if parent_row is None:
+                raise ResourceNotFoundRepoError(resource_id)
+
+            parent_type_name = cast(str, parent_row.type_name)
+            type_row = (
+                session.query(ResourceTypeModel)
+                .filter_by(name=parent_type_name)
+                .first()
+            )
+            if type_row is None:
+                return []
+
+            # Parse auto_discovery config
+            auto_disc_raw = cast("str | None", type_row.auto_discover_json)
+            if not auto_disc_raw:
+                return []
+
+            auto_disc: dict[str, Any] = json.loads(auto_disc_raw)
+            if not auto_disc.get("enabled", False):
+                return []
+
+            rules: list[dict[str, Any]] = auto_disc.get("rules", [])
+            if not rules:
+                return []
+
+            # Parse allowed child types
+            child_types_raw = cast(
+                "str | None",
+                type_row.allowed_child_types_json,
+            )
+            allowed_child_types: list[str] = (
+                json.loads(child_types_raw) if child_types_raw else []
+            )
+
+            created: list[Any] = []
+            for rule in rules:
+                child_type_name = rule.get("type", "")
+                if not child_type_name:
+                    continue
+
+                # Verify child type exists in DB
+                ct_row = (
+                    session.query(ResourceTypeModel)
+                    .filter_by(name=child_type_name)
+                    .first()
+                )
+                if ct_row is None:
+                    continue
+
+                # Verify type compatibility
+                if allowed_child_types and child_type_name not in allowed_child_types:
+                    continue
+
+                ct_kind = cast(str, ct_row.resource_kind)
+
+                now_iso = datetime.now(tz=UTC).isoformat()
+                child_id = str(_ULID())
+                child_model = ResourceModel(
+                    resource_id=child_id,
+                    namespaced_name=None,
+                    namespace=None,
+                    type_name=child_type_name,
+                    resource_kind=ct_kind,
+                    location=None,
+                    description=(f"Auto-discovered {child_type_name}"),
+                    read_only=False,
+                    auto_discovered=True,
+                    sandbox_strategy=None,
+                    content_hash=None,
+                    properties_json=None,
+                    metadata_json=None,
+                    created_at=now_iso,
+                    updated_at=now_iso,
+                )
+                session.add(child_model)
+                session.flush()
+
+                # Link child to parent
+                link = ResourceLinkModel(
+                    parent_id=resource_id,
+                    child_id=child_id,
+                    created_at=now_iso,
+                )
+                session.add(link)
+                session.flush()
+
+                child_resource = Resource(
+                    resource_id=child_id,
+                    name=None,
+                    resource_type_name=child_type_name,
+                    classification=PhysVirt(ct_kind),
+                    description=(f"Auto-discovered {child_type_name}"),
+                    properties={},
+                    location=None,
+                    content_hash=None,
+                    sandbox_strategy=None,
+                    capabilities=ResourceCapabilities(),
+                    created_at=datetime.fromisoformat(now_iso),
+                    updated_at=datetime.fromisoformat(now_iso),
+                )
+                created.append(child_resource)
+
+            return created
+        except ResourceNotFoundRepoError:
+            raise
+        except (
+            OperationalError,
+            SQLAlchemyDatabaseError,
+        ) as exc:
+            session.rollback()
+            raise DatabaseError(
+                f"Failed to auto-discover children for '{resource_id}': {exc}"
+            ) from exc
+
+    @staticmethod
+    def _get_ancestors(session: Session, resource_id: str) -> set[str]:
+        """Return all ancestor resource IDs (BFS upward).
+
+        Used for cycle detection: if a proposed child is
+        already an ancestor of the parent, linking would
+        create a cycle.
+        """
+        visited: set[str] = set()
+        queue: list[str] = [resource_id]
+        while queue:
+            current = queue.pop(0)
+            if current in visited:
+                continue
+            visited.add(current)
+            parent_links = (
+                session.query(ResourceLinkModel).filter_by(child_id=current).all()
+            )
+            for link in parent_links:
+                pid = cast(str, link.parent_id)
+                if pid not in visited:
+                    queue.append(pid)
+        return visited
+
+    @staticmethod
+    def _build_cycle_path(
+        session: Session,
+        parent_id: str,
+        child_id: str,
+    ) -> list[str]:
+        """Build a path showing the cycle for error msgs.
+
+        Returns a list like [child_id, ..., parent_id,
+        child_id] showing the cycle.
+        """
+        # BFS from child_id upward to find parent_id
+        predecessors: dict[str, str | None] = {parent_id: None}
+        queue: list[str] = [parent_id]
+        found = False
+        while queue and not found:
+            current = queue.pop(0)
+            parent_links = (
+                session.query(ResourceLinkModel).filter_by(child_id=current).all()
+            )
+            for link in parent_links:
+                pid = cast(str, link.parent_id)
+                if pid not in predecessors:
+                    predecessors[pid] = current
+                    if pid == child_id:
+                        found = True
+                        break
+                    queue.append(pid)
+
+        # Reconstruct path
+        path: list[str] = []
+        current_node: str | None = child_id
+        while current_node is not None:
+            path.append(current_node)
+            current_node = predecessors.get(current_node)
+        path.append(child_id)
+        return path
+
     @staticmethod
     def _to_domain(row: ResourceModel) -> Any:
         """Convert a ``ResourceModel`` row to a ``Resource`` domain object."""

vulture_whitelist.py

@@ -114,3 +114,17 @@ AutomationProfileSchemaVersionError  # noqa: B018, F821
 _LEVEL_TO_PROFILE  # noqa: B018, F821
 _resolve_profile_for_plan  # noqa: B018, F821
 default_automation_profile  # noqa: B018, F821
+
+# Resource DAG — public API and model attributes
+ResourceLinkModel  # noqa: B018, F821
+CycleDetectedError  # noqa: B018, F821
+TypeIncompatibleError  # noqa: B018, F821
+LinkNotFoundError  # noqa: B018, F821
+DuplicateResourceLinkError  # noqa: B018, F821
+link_child  # noqa: B018, F821
+unlink_child  # noqa: B018, F821
+get_children  # noqa: B018, F821
+get_parents  # noqa: B018, F821
+auto_discover_children  # noqa: B018, F821
+_get_ancestors  # noqa: B018, F821
+_build_cycle_path  # noqa: B018, F821