feat(resource): add virtual resource equivalence tracking

Virtual-to-physical resource equivalence linking system for tracking
which physical resources are manifestations of the same virtual identity.

Database:
- VirtualResourceLinkModel ORM with ULID PK, FK cascades, unique
  constraint on (virtual_resource_id, physical_resource_id), CHECK
  constraints on link_type and confidence, 4 indexes
- Alembic migration: m7_001_virtual_resource_links.py (depends on
  m8_002_merge_profile_rename_and_corrections)

Domain model:
- LinkType enum (content_hash, name, identity)
- VirtualResourceLink frozen Pydantic model with confidence [0.0-1.0]

Service (ResourceEquivalenceService — 498 lines):
- create_link() with type-mismatch guard via ALLOWED_PAIRINGS
- remove_link() by virtual+physical ID pair
- list_links() with pagination (limit/offset)
- merge_virtual_resources() — batch N+1-free with IntegrityError guard
- check_divergence() — removes stale content_hash links
- reconcile() — delegated to _resource_equivalence_reconciler.py

Helpers (_resource_equivalence_helpers.py — 228 lines):
- compute_equivalence_key() for string content (SHA-256/name/identity)
- compute_equivalence_key_from_path() streaming file hash (64 KiB chunks)
- ALLOWED_PAIRINGS type compatibility map (unprefixed virtual type names)
- validate_link_inputs() shared validation

CLI (resource_equivalence.py):
- agents resource equivalence list (--virtual, --physical, --format)
- agents resource equivalence add (--key, --type, --confidence, --format)
- agents resource equivalence remove (--yes, --format)

Tests:
- 50 Behave scenarios (CRUD, merge, divergence, reconciliation, boundaries)
- 9 Robot integration tests
- 5 ASV benchmark suites

ISSUES CLOSED: #334

CHANGELOG.md

@@ -273,6 +273,12 @@
   fragments to overview depths 0-1 using the UKO detail-level map chain,
   exposes the compressor as the configured builtin, and covers the behavior
   with BDD scenarios for compressor output and default pipeline wiring. (#919)
+- Added virtual resource equivalence tracking system.
+  `VirtualResourceLink` frozen Pydantic domain model and `LinkType` StrEnum
+  (`content_hash`, `name`, `identity`). `ResourceEquivalenceService` with
+  CRUD, merge, content-hash-scoped divergence detection, and batch
+  reconciliation. CLI commands `agents resource equivalence list/add/remove`.
+  (#334)
 - Added TDD bug-capture tests for bug #1076 — `use_action()` does not
   propagate `automation_profile` to Plan. Three Behave BDD scenarios
   (`@tdd_bug @tdd_bug_1076 @tdd_expected_fail`) verify the full precedence

alembic/versions/m7_001_virtual_resource_links.py

@@ -0,0 +1,104 @@
+"""Add virtual_resource_links table for equivalence tracking.
+
+Revision ID: m7_001_virtual_resource_links
+Revises: m8_002_merge_profile_rename_and_corrections
+Create Date: 2026-03-10 00:00:00
+
+"""
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision: str = "m7_001_virtual_resource_links"
+down_revision: str | Sequence[str] | None = (
+    "m8_002_merge_profile_rename_and_corrections"  # noqa: E501
+)
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    """Create the virtual_resource_links table."""
+    op.create_table(
+        "virtual_resource_links",
+        sa.Column("id", sa.String(26), primary_key=True),
+        sa.Column(
+            "virtual_resource_id",
+            sa.String(26),
+            sa.ForeignKey("resources.resource_id", ondelete="CASCADE"),
+            nullable=False,
+        ),
+        sa.Column(
+            "physical_resource_id",
+            sa.String(26),
+            sa.ForeignKey("resources.resource_id", ondelete="CASCADE"),
+            nullable=False,
+        ),
+        sa.Column("equivalence_key", sa.String(1024), nullable=False),
+        sa.Column(
+            "link_type",
+            sa.String(30),
+            nullable=False,
+            server_default="content_hash",
+        ),
+        sa.Column("confidence", sa.Float(), nullable=False, server_default="1.0"),
+        sa.Column("created_at", sa.String(40), nullable=False),
+        sa.Column("updated_at", sa.String(40), nullable=False),
+        sa.UniqueConstraint(
+            "virtual_resource_id",
+            "physical_resource_id",
+            name="uq_virtual_resource_links_virtual_physical",
+        ),
+        sa.CheckConstraint(
+            "link_type IN ('content_hash', 'name', 'identity')",
+            name="ck_virtual_resource_links_link_type",
+        ),
+        sa.CheckConstraint(
+            "confidence >= 0.0 AND confidence <= 1.0",
+            name="ck_virtual_resource_links_confidence",
+        ),
+    )
+    op.create_index(
+        "ix_virtual_resource_links_virtual",
+        "virtual_resource_links",
+        ["virtual_resource_id"],
+    )
+    op.create_index(
+        "ix_virtual_resource_links_physical",
+        "virtual_resource_links",
+        ["physical_resource_id"],
+    )
+    op.create_index(
+        "ix_virtual_resource_links_equiv_key",
+        "virtual_resource_links",
+        ["equivalence_key"],
+    )
+    op.create_index(
+        "ix_virtual_resource_links_link_type",
+        "virtual_resource_links",
+        ["link_type"],
+    )
+
+
+def downgrade() -> None:
+    """Drop the virtual_resource_links table."""
+    op.drop_index(
+        "ix_virtual_resource_links_link_type",
+        table_name="virtual_resource_links",
+    )
+    op.drop_index(
+        "ix_virtual_resource_links_equiv_key",
+        table_name="virtual_resource_links",
+    )
+    op.drop_index(
+        "ix_virtual_resource_links_physical",
+        table_name="virtual_resource_links",
+    )
+    op.drop_index(
+        "ix_virtual_resource_links_virtual",
+        table_name="virtual_resource_links",
+    )
+    op.drop_table("virtual_resource_links")

benchmarks/virtual_resource_bench.py

@@ -0,0 +1,286 @@
+"""ASV benchmarks for virtual resource equivalence tracking.
+
+Benchmarks cover:
+- compute_equivalence_key (content_hash / name / identity)
+- ResourceEquivalenceService.create_link
+- ResourceEquivalenceService.list_links
+- ResourceEquivalenceService.check_divergence
+- ResourceEquivalenceService.reconcile
+"""
+
+from __future__ import annotations
+
+import typing
+from datetime import UTC, datetime
+from typing import Any
+
+_CB32 = "0123456789ABCDEFGHJKMNPQRSTVWXYZ"
+_BENCH_CTR = 0
+
+
+def _bench_ulid() -> str:
+    global _BENCH_CTR
+    _BENCH_CTR += 1
+    n = _BENCH_CTR
+    suffix = ""
+    for _ in range(7):
+        suffix = _CB32[n % 32] + suffix
+        n //= 32
+    return f"01HEQVBNCH0AQDYTR4B{suffix}"[:26]
+
+
+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)
+
+
+def _seed_types(session_factory: Any) -> None:
+    """Register virtual and physical resource types."""
+    from cleveragents.infrastructure.database.models import ResourceTypeModel
+
+    now_iso = datetime.now(tz=UTC).isoformat()
+    session = session_factory()
+    session.add(
+        ResourceTypeModel(
+            name="file",
+            namespace="builtin",
+            description="Virtual file",
+            resource_kind="virtual",
+            sandbox_strategy="none",
+            user_addable=True,
+            args_schema_json="[]",
+            allowed_parent_types_json="[]",
+            allowed_child_types_json="[]",
+            capabilities_json='{"read":true,"write":false,"sandbox":false,"checkpoint":false}',
+            source="builtin",
+            created_at=now_iso,
+            updated_at=now_iso,
+        )
+    )
+    session.add(
+        ResourceTypeModel(
+            name="fs-file",
+            namespace="builtin",
+            description="File system file",
+            resource_kind="physical",
+            sandbox_strategy="none",
+            user_addable=True,
+            args_schema_json="[]",
+            allowed_parent_types_json="[]",
+            allowed_child_types_json="[]",
+            capabilities_json='{"read":true,"write":true,"sandbox":true,"checkpoint":false}',
+            source="builtin",
+            created_at=now_iso,
+            updated_at=now_iso,
+        )
+    )
+    session.commit()
+    session.close()
+
+
+def _make_resource(
+    session_factory: Any,
+    *,
+    resource_kind: str = "physical",
+    type_name: str = "fs-file",
+    content_hash: str | None = None,
+) -> str:
+    """Create a resource row and return its ULID."""
+    from cleveragents.infrastructure.database.models import ResourceModel
+
+    session = session_factory()
+    rid = _bench_ulid()
+    now_iso = datetime.now(tz=UTC).isoformat()
+    session.add(
+        ResourceModel(
+            resource_id=rid,
+            type_name=type_name,
+            resource_kind=resource_kind,
+            content_hash=content_hash,
+            created_at=now_iso,
+            updated_at=now_iso,
+        )
+    )
+    session.commit()
+    session.close()
+    return rid
+
+
+# ---------------------------------------------------------------------------
+# Benchmark: compute_equivalence_key
+# ---------------------------------------------------------------------------
+
+
+class TimeComputeEquivalenceKey:
+    """Benchmark compute_equivalence_key for each link type."""
+
+    params: typing.ClassVar[list[str]] = ["content_hash", "name", "identity"]
+    param_names: typing.ClassVar[list[str]] = ["link_type"]
+
+    def setup(self, link_type: str) -> None:
+        from cleveragents.application.services.resource_equivalence_service import (
+            compute_equivalence_key,
+        )
+
+        self._compute = compute_equivalence_key
+        self._content = "x" * 4096
+        self._name = "main-branch"
+
+    def time_compute(self, link_type: str) -> None:
+        if link_type == "content_hash":
+            self._compute(content=self._content, link_type=link_type)
+        else:
+            self._compute(name=self._name, link_type=link_type)
+
+
+# ---------------------------------------------------------------------------
+# Benchmark: create_link
+# ---------------------------------------------------------------------------
+
+
+class TimeCreateLink:
+    """Benchmark ResourceEquivalenceService.create_link."""
+
+    def setup(self) -> None:
+        from cleveragents.application.services.resource_equivalence_service import (
+            ResourceEquivalenceService,
+        )
+
+        self.factory = _setup_db()
+        _seed_types(self.factory)
+        self.svc = ResourceEquivalenceService(self.factory)
+        self._counter = 0
+
+    def time_create_link(self) -> None:
+        self._counter += 1
+        v_id = _make_resource(self.factory, resource_kind="virtual", type_name="file")
+        p_id = _make_resource(
+            self.factory,
+            resource_kind="physical",
+            type_name="fs-file",
+            content_hash=f"hash_{self._counter}",
+        )
+        self.svc.create_link(v_id, p_id, f"hash_{self._counter}")
+
+
+# ---------------------------------------------------------------------------
+# Benchmark: list_links
+# ---------------------------------------------------------------------------
+
+
+class TimeListLinks:
+    """Benchmark ResourceEquivalenceService.list_links with N links."""
+
+    params: typing.ClassVar[list[int]] = [1, 10, 50]
+    param_names: typing.ClassVar[list[str]] = ["num_links"]
+
+    def setup(self, num_links: int) -> None:
+        from cleveragents.application.services.resource_equivalence_service import (
+            ResourceEquivalenceService,
+        )
+
+        self.factory = _setup_db()
+        _seed_types(self.factory)
+        self.svc = ResourceEquivalenceService(self.factory)
+
+        self.virtual_id = _make_resource(
+            self.factory, resource_kind="virtual", type_name="file"
+        )
+        for i in range(num_links):
+            p_id = _make_resource(
+                self.factory,
+                resource_kind="physical",
+                type_name="fs-file",
+                content_hash=f"hash_{i}",
+            )
+            self.svc.create_link(self.virtual_id, p_id, f"hash_{i}")
+
+    def time_list_links(self, num_links: int) -> None:
+        self.svc.list_links(virtual_resource_id=self.virtual_id)
+
+
+# ---------------------------------------------------------------------------
+# Benchmark: check_divergence
+# ---------------------------------------------------------------------------
+
+
+class TimeCheckDivergence:
+    """Benchmark check_divergence with varying link counts."""
+
+    params: typing.ClassVar[list[int]] = [1, 5, 20]
+    param_names: typing.ClassVar[list[str]] = ["num_links"]
+
+    def setup(self, num_links: int) -> None:
+        from cleveragents.application.services.resource_equivalence_service import (
+            ResourceEquivalenceService,
+        )
+
+        self.factory = _setup_db()
+        _seed_types(self.factory)
+        self.svc = ResourceEquivalenceService(self.factory)
+
+        # Create one physical resource linked to many virtual resources
+        self.physical_id = _make_resource(
+            self.factory,
+            resource_kind="physical",
+            type_name="fs-file",
+            content_hash="original_hash",
+        )
+        for _i in range(num_links):
+            v_id = _make_resource(
+                self.factory, resource_kind="virtual", type_name="file"
+            )
+            self.svc.create_link(v_id, self.physical_id, "original_hash")
+
+    def time_check_divergence(self, num_links: int) -> None:
+        # Check with same key (no divergence) — re-runnable
+        self.svc.check_divergence(self.physical_id, "original_hash")
+
+
+# ---------------------------------------------------------------------------
+# Benchmark: reconcile
+# ---------------------------------------------------------------------------
+
+
+class TimeReconcile:
+    """Benchmark reconcile with varying link counts."""
+
+    params: typing.ClassVar[list[int]] = [5, 20, 50]
+    param_names: typing.ClassVar[list[str]] = ["num_links"]
+
+    def setup(self, num_links: int) -> None:
+        from cleveragents.application.services.resource_equivalence_service import (
+            ResourceEquivalenceService,
+        )
+
+        self.factory = _setup_db()
+        _seed_types(self.factory)
+        self.svc = ResourceEquivalenceService(self.factory)
+
+        # Create links where content_hash matches (so reconcile keeps them)
+        for i in range(num_links):
+            v_id = _make_resource(
+                self.factory, resource_kind="virtual", type_name="file"
+            )
+            p_id = _make_resource(
+                self.factory,
+                resource_kind="physical",
+                type_name="fs-file",
+                content_hash=f"hash_{i}",
+            )
+            self.svc.create_link(v_id, p_id, f"hash_{i}")
+
+    def time_reconcile(self, num_links: int) -> None:
+        self.svc.reconcile()

docs/reference/resource_model.md

@@ -0,0 +1,176 @@
+# Resource Equivalence Tracking
+
+The **Resource Equivalence** system tracks the relationship between virtual
+and physical resources in the CleverAgents resource DAG.  A virtual resource
+represents an abstract identity (e.g., "the file `README.md`"), while a
+physical resource represents a concrete manifestation (e.g., the actual
+`README.md` in the working tree at `/project/README.md`).
+
+---
+
+## Overview
+
+| Component | Role |
+|-----------|------|
+| `VirtualResourceLink` | Frozen Pydantic domain model for a single equivalence link |
+| `LinkType` | `StrEnum` of equivalence methods (`content_hash`, `name`, `identity`) |
+| `ResourceEquivalenceService` | Application service for CRUD, merge, divergence, reconciliation |
+| `compute_equivalence_key()` | Helper to compute hash/name/identity keys |
+| `VirtualResourceLinkModel` | SQLAlchemy ORM model for the `virtual_resource_links` table |
+
+---
+
+## Link Types
+
+| Value | Description | Example |
+|-------|-------------|---------|
+| `content_hash` | SHA-256 of file content | Two files with identical bytes |
+| `name` | Normalised name (lowercase, stripped) | Branch name `main` |
+| `identity` | Opaque identity value | Commit hash `abc123` |
+
+---
+
+## Equivalence Key Computation
+
+```python
+from cleveragents.application.services import compute_equivalence_key
+
+# Content-based (SHA-256)
+key = compute_equivalence_key(content="hello world", link_type="content_hash")
+# -> "b94d27b9934d3e08a52e52d7da7dabfac484efe37a5380ee9088f7ace2efcde9"
+
+# Name-based (normalised)
+key = compute_equivalence_key(name="  Main  ", link_type="name")
+# -> "main"
+
+# Identity-based (pass-through)
+key = compute_equivalence_key(name="abc123def", link_type="identity")
+# -> "abc123def"
+```
+
+---
+
+## Type Mismatch Guard
+
+Virtual types can only link to compatible physical types:
+
+| Virtual Type | Allowed Physical Types |
+|-------------|----------------------|
+| `file` | `fs-file`, `git-tree-entry` |
+| `directory` | `fs-directory`, `git-tree` |
+| `commit` | `git-commit` |
+| `branch` | `git-branch` |
+| `tag` | `git-tag` |
+| `remote` | `git-remote` |
+| `submodule` | `git-submodule` |
+| `tree` | `git-tree` |
+| `symlink` | `fs-symlink`, `git-tree-entry` |
+
+User-defined virtual types (not in this table) are allowed to link to any
+physical type.
+
+---
+
+## Conflict Policy
+
+Physical resources take precedence over virtual resources.  When a physical
+resource's content changes (detected via `check_divergence` or `reconcile`),
+the equivalence link is removed and a divergence event is logged.  Virtual
+resources are never mutated directly — they serve as read-only identity
+anchors.
+
+---
+
+## Database Schema
+
+The `virtual_resource_links` table:
+
+| Column | Type | Constraints |
+|--------|------|-------------|
+| `id` | `TEXT` | Primary key (ULID) |
+| `virtual_resource_id` | `TEXT` | FK → `resources.resource_id`, NOT NULL |
+| `physical_resource_id` | `TEXT` | FK → `resources.resource_id`, NOT NULL |
+| `equivalence_key` | `VARCHAR(1024)` | NOT NULL |
+| `link_type` | `TEXT` | CHECK IN (`content_hash`, `name`, `identity`) |
+| `confidence` | `REAL` | CHECK 0.0 ≤ confidence ≤ 1.0, DEFAULT 1.0 |
+| `created_at` | `TEXT` | NOT NULL |
+| `updated_at` | `TEXT` | NOT NULL |
+
+**Unique constraint**: `(virtual_resource_id, physical_resource_id)`
+
+**Indexes**:
+- `ix_virtual_resource_links_virtual` on `virtual_resource_id`
+- `ix_virtual_resource_links_physical` on `physical_resource_id`
+- `ix_virtual_resource_links_equiv_key` on `equivalence_key`
+- `ix_virtual_resource_links_link_type` on `link_type`
+
+---
+
+## CLI Commands
+
+### List equivalence links
+
+```bash
+# List all links for a virtual resource
+agents resource equivalence list --virtual 01HXYZ...
+
+# List all links for a physical resource
+agents resource equivalence list --physical 01HABC...
+
+# List all links
+agents resource equivalence list
+```
+
+### Add an equivalence link
+
+```bash
+agents resource equivalence add 01HXYZ... 01HABC... \
+  --key "b94d27b993..." \
+  --type content_hash \
+  --confidence 1.0
+```
+
+### Remove an equivalence link
+
+```bash
+agents resource equivalence remove 01HXYZ... 01HABC...
+```
+
+---
+
+## Service API
+
+```python
+from cleveragents.application.services import ResourceEquivalenceService
+
+svc = ResourceEquivalenceService(session_factory)
+
+# Create a link
+link = svc.create_link(virtual_id, physical_id, equiv_key,
+                        link_type="content_hash", confidence=1.0)
+
+# List links
+links = svc.list_links(virtual_resource_id=virtual_id)
+
+# Remove a link
+svc.remove_link(virtual_id, physical_id)
+
+# Merge two virtual resources
+merged = svc.merge_virtual_resources(target_id, source_id)
+
+# Check divergence when physical content changes
+remaining = svc.check_divergence(physical_id, new_equiv_key)
+
+# Reconciliation job (re-hash all content_hash links)
+result = svc.reconcile()  # {"checked": 10, "removed": 2, "kept": 8}
+```
+
+---
+
+## Specification References
+
+- Physical vs Virtual Resources: `specification.md` lines 24383-24435
+- Lazy Virtual Node Materialization: `specification.md` lines 24437-24468
+- Equivalence Linking: `specification.md` lines 24397-24417
+- Divergence Detection: `specification.md` lines 24419-24435
+- Issue: [#334](https://git.cleverthis.com/cleveragents/cleveragents-core/issues/334)

features/consolidated_domain_models.feature

@@ -1108,7 +1108,7 @@ Feature: Consolidated Domain Models
 
   Scenario: CLI dict has required fields
     Given a session with some messages
-    When I get the session CLI dict
+    When I get the session model CLI dict
     Then the session cli dict should have key "session_id"
     And the session cli dict should have key "message_count"
     And the session cli dict should have key "created_at"
@@ -1119,13 +1119,13 @@ Feature: Consolidated Domain Models
 
   Scenario: CLI dict includes recent messages
     Given a session with some messages
-    When I get the session CLI dict
+    When I get the session model CLI dict
     Then the session cli dict should have key "recent_messages"
 
 
   Scenario: CLI dict includes actor when set
     When I create a session with actor name "local/orchestrator"
-    And I get the session CLI dict
+    And I get the session model CLI dict
     Then the session cli dict should have key "actor_name"
 
     # ---- Empty Session Properties ----

features/edge_case_plan_scenarios.feature

@@ -180,19 +180,19 @@ Feature: Plan edge case scenarios
 
   Scenario: Plan model rejects empty description
     When I try to create an edge case plan with empty description
-    Then a Pydantic validation error should be raised
+    Then a Pydantic validation error should be raised for edge case plan
 
   Scenario: Plan model rejects invalid phase value
     When I try to create an edge case plan with invalid phase value
-    Then a Pydantic validation error should be raised
+    Then a Pydantic validation error should be raised for edge case plan
 
   Scenario: NamespacedName rejects invalid characters in namespace
     When I try to parse a namespaced name with special characters "inv@lid/action"
-    Then a Pydantic validation error should be raised
+    Then a Pydantic validation error should be raised for edge case plan
 
   Scenario: NamespacedName rejects invalid characters in name
     When I try to parse a namespaced name with special chars in name "local/my action!"
-    Then a Pydantic validation error should be raised
+    Then a Pydantic validation error should be raised for edge case plan
 
   # ──────────────────────────────────────────────────
   # Section 4: Rollback edge cases

features/plan_executor_coverage_boost.feature

@@ -32,7 +32,7 @@ Feature: PlanExecutor Coverage Boost
   Scenario: _try_rollback_to_last_checkpoint returns False when no sandbox is resolvable
     Given a PlanExecutor with a checkpoint manager but no sandbox source
     When I attempt to rollback to the last checkpoint
-    Then the rollback result should be False
+    Then the rollback result should be False for executor
 
   # --- _resolve_sandbox_for_checkpoint via execution_context (line 458) ---
 

features/plan_explain.feature

@@ -56,7 +56,7 @@ Feature: Plan explain and decision tree CLI commands
     Given a test decision for explain
     When I format the explain dict as json
     Then the json output should contain "decision_id"
-    And the json output should be valid json
+    And the plan explain json output should be valid json
 
   # ------------------------------------------------------------------
   # plan explain - yaml format

features/project_repository.feature

@@ -125,7 +125,7 @@ Feature: Namespaced project repository operations
 
   Scenario: Remove a non-existent link returns False
     When I remove a link with id "00000000000000000000000099"
-    Then the remove result should be False
+    Then the project remove result should be False
 
   Scenario: Create link with read_only flag
     Given a namespaced project "local/ro-proj" exists in the repository

features/resource_cli_flags_904.feature

@@ -35,7 +35,7 @@ Feature: Resource and LSP CLI missing flags (issue #904)
     And resource flags built-in types are bootstrapped
     When I run resource flags add "git-checkout" "local/bad-clone" with path "/tmp/bc" and clone-into "https://example.com/repo.git:/workspace"
     Then the resource flags command should fail
-    And the resource flags output should contain "container types"
+    And the resource flags output should contain "container resource types"
 
   Scenario: Clone-into flag with invalid format
     Given a fresh resource flags test registry

features/resource_equivalence.feature

@@ -0,0 +1,282 @@
+Feature: Virtual Resource Equivalence Tracking
+  As a CleverAgents user
+  I want to track equivalence between virtual and physical resources
+  So that I can discover where the same content exists across the resource DAG
+
+  Background:
+    Given a fresh in-memory database for equiv
+    And a virtual resource type registered for equiv
+    And a physical resource type registered for equiv
+
+  # -- Domain model tests ---------------------------------------------------
+
+  Scenario: VirtualResourceLink model has all spec fields for equiv
+    When I create a VirtualResourceLink domain model for equiv
+    Then the link model has fields id, virtual_resource_id, physical_resource_id for equiv
+    And the link model has fields equivalence_key, link_type, confidence for equiv
+    And the link model has fields created_at, updated_at for equiv
+
+  Scenario: VirtualResourceLink model is frozen for equiv
+    When I create a VirtualResourceLink domain model for equiv
+    Then the link model is immutable for equiv
+
+  Scenario: LinkType enum has all required values for equiv
+    Then LinkType has values content_hash, name, identity for equiv
+
+  Scenario: VirtualResourceLink rejects empty equivalence_key for equiv
+    Then creating a link with empty equivalence_key raises ValueError for equiv
+
+  Scenario: VirtualResourceLink rejects confidence below 0.0 for equiv
+    Then creating a link with confidence -0.1 raises ValueError for equiv
+
+  Scenario: VirtualResourceLink rejects confidence above 1.0 for equiv
+    Then creating a link with confidence 1.1 raises ValueError for equiv
+
+  # -- Equivalence key computation ------------------------------------------
+
+  Scenario: Compute content_hash equivalence key for equiv
+    When I compute an equivalence key with content "hello world" and type content_hash for equiv
+    Then the equivalence key is a SHA-256 hex digest for equiv
+
+  Scenario: Compute name equivalence key for equiv
+    When I compute an equivalence key with name "Main Branch" and type name for equiv
+    Then the equivalence key is "main branch" for equiv
+
+  Scenario: Compute identity equivalence key for equiv
+    When I compute an equivalence key with name "abc123def" and type identity for equiv
+    Then the equivalence key is "abc123def" for equiv
+
+  Scenario: Content hash key requires content for equiv
+    Then computing an equivalence key without content for content_hash raises ValidationError for equiv
+
+  Scenario: Name key requires name for equiv
+    Then computing an equivalence key without name for name type raises ValidationError for equiv
+
+  Scenario: Identity key requires name for equiv
+    Then computing an equivalence key without name for identity type raises ValidationError for equiv
+
+  Scenario: Unknown link type raises error for equiv
+    Then computing an equivalence key with unknown type raises ValidationError for equiv
+
+  # -- Service CRUD ---------------------------------------------------------
+
+  Scenario: Create an equivalence link for equiv
+    Given a virtual resource exists for equiv
+    And a physical resource exists for equiv
+    When I create an equivalence link between them for equiv
+    Then the link is persisted with correct fields for equiv
+
+  Scenario: List equivalence links for equiv
+    Given a virtual resource exists for equiv
+    And a physical resource exists for equiv
+    And an equivalence link exists between them for equiv
+    When I list equivalence links for equiv
+    Then the list contains 1 link for equiv
+
+  Scenario: List links filtered by virtual resource for equiv
+    Given a virtual resource exists for equiv
+    And a physical resource exists for equiv
+    And an equivalence link exists between them for equiv
+    When I list equivalence links filtered by virtual resource for equiv
+    Then the list contains 1 link for equiv
+
+  Scenario: List links filtered by physical resource for equiv
+    Given a virtual resource exists for equiv
+    And a physical resource exists for equiv
+    And an equivalence link exists between them for equiv
+    When I list equivalence links filtered by physical resource for equiv
+    Then the list contains 1 link for equiv
+
+  Scenario: Remove an equivalence link for equiv
+    Given a virtual resource exists for equiv
+    And a physical resource exists for equiv
+    And an equivalence link exists between them for equiv
+    When I remove the equivalence link for equiv
+    Then the link no longer exists for equiv
+
+  Scenario: Remove nonexistent link raises NotFoundError for equiv
+    Then removing a nonexistent equivalence link raises NotFoundError for equiv
+
+  Scenario: Create duplicate link raises ValidationError for equiv
+    Given a virtual resource exists for equiv
+    And a physical resource exists for equiv
+    And an equivalence link exists between them for equiv
+    Then creating a duplicate equivalence link raises ValidationError for equiv
+
+  # -- Type mismatch guard --------------------------------------------------
+
+  Scenario: Linking incompatible types raises ValidationError for equiv
+    Given a virtual resource of type "file" exists for equiv
+    And a physical resource of type "git-commit" exists for equiv
+    Then creating a link between incompatible types raises ValidationError for equiv
+
+  # -- Physical must be physical, virtual must be virtual -------------------
+
+  Scenario: Linking two physical resources raises ValidationError for equiv
+    Given two physical resources exist for equiv
+    Then creating a link with physical as virtual raises ValidationError for equiv
+
+  Scenario: Linking two virtual resources raises ValidationError for equiv
+    Given two virtual resources exist for equiv
+    Then creating a link with virtual as physical raises ValidationError for equiv
+
+  # -- Merge virtual resources -----------------------------------------------
+
+  Scenario: Merge two virtual resources for equiv
+    Given two virtual resources with physical links exist for equiv
+    When I merge the source virtual into the target virtual for equiv
+    Then the target virtual has all physical links for equiv
+
+  # -- Divergence detection -------------------------------------------------
+
+  Scenario: Divergence removes stale links for equiv
+    Given a virtual resource exists for equiv
+    And a physical resource exists for equiv
+    And an equivalence link exists between them for equiv
+    When I check divergence with a new key for equiv
+    Then the stale link is removed for equiv
+
+  Scenario: Non-diverged links remain after check for equiv
+    Given a virtual resource exists for equiv
+    And a physical resource exists for equiv
+    And an equivalence link exists between them for equiv
+    When I check divergence with the same key for equiv
+    Then the link still exists for equiv
+
+  # -- Reconciliation job ---------------------------------------------------
+
+  Scenario: Reconciliation removes stale hash links for equiv
+    Given a virtual resource exists for equiv
+    And a physical resource with content_hash exists for equiv
+    And an equivalence link with mismatched hash exists for equiv
+    When I run the reconciliation job for equiv
+    Then the stale link is removed by reconciliation for equiv
+
+  Scenario: Reconciliation keeps valid hash links for equiv
+    Given a virtual resource exists for equiv
+    And a physical resource with content_hash exists for equiv
+    And an equivalence link with matching hash exists for equiv
+    When I run the reconciliation job for equiv
+    Then the valid link is kept by reconciliation for equiv
+
+  # -- Empty equivalence key validation for equiv ---------------------------
+
+  Scenario: Service rejects empty equivalence key for equiv
+    Given a virtual resource exists for equiv
+    And a physical resource exists for equiv
+    Then creating a link with empty key via service raises ValidationError for equiv
+
+  # -- Conflict policy: physical preferred over virtual ---------------------
+
+  Scenario: Physical resource change unlinks from virtual parent for equiv
+    Given a virtual resource exists for equiv
+    And a physical resource exists for equiv
+    And an equivalence link exists between them for equiv
+    When the physical resource content changes for equiv
+    Then the physical resource is unlinked from its virtual parent for equiv
+
+  # -- Boundary condition tests (TEST-1) ------------------------------------
+
+  Scenario: Service rejects whitespace-only equivalence key for equiv
+    Given a virtual resource exists for equiv
+    And a physical resource exists for equiv
+    Then creating a link with whitespace key via service raises ValidationError for equiv
+
+  Scenario: Service rejects invalid link_type for equiv
+    Given a virtual resource exists for equiv
+    And a physical resource exists for equiv
+    Then creating a link with invalid link_type via service raises ValidationError for equiv
+
+  Scenario: Confidence 0.0 is valid for equiv
+    Given a virtual resource exists for equiv
+    And a physical resource exists for equiv
+    When I create an equivalence link with confidence 0.0 for equiv
+    Then the created link has confidence 0.0 for equiv
+
+  Scenario: Confidence 1.0 is valid for equiv
+    Given a virtual resource exists for equiv
+    And a physical resource exists for equiv
+    When I create an equivalence link with confidence 1.0 for equiv
+    Then the created link has confidence 1.0 for equiv
+
+  Scenario: Confidence -0.1 is rejected by service for equiv
+    Given a virtual resource exists for equiv
+    And a physical resource exists for equiv
+    Then creating a link with confidence -0.1 via service raises ValidationError for equiv
+
+  Scenario: Confidence 1.1 is rejected by service for equiv
+    Given a virtual resource exists for equiv
+    And a physical resource exists for equiv
+    Then creating a link with confidence 1.1 via service raises ValidationError for equiv
+
+  Scenario: Nonexistent virtual resource raises NotFoundError for equiv
+    Given a physical resource exists for equiv
+    Then creating a link with nonexistent virtual resource raises NotFoundError for equiv
+
+  Scenario: Nonexistent physical resource raises NotFoundError for equiv
+    Given a virtual resource exists for equiv
+    Then creating a link with nonexistent physical resource raises NotFoundError for equiv
+
+  Scenario: Check divergence rejects empty key for equiv
+    Then checking divergence with empty key raises ValidationError for equiv
+
+  Scenario: Name equivalence key strips and lowercases for equiv
+    When I compute an equivalence key with name "  UPPER  " and type name for equiv
+    Then the equivalence key is "upper" for equiv
+
+  Scenario: Name equivalence key rejects whitespace-only for equiv
+    Then computing a name key with whitespace-only raises ValidationError for equiv
+
+  Scenario: Identity key preserves case for equiv
+    When I compute an equivalence key with name "AbCdEf" and type identity for equiv
+    Then the equivalence key is "AbCdEf" for equiv
+
+  # -- Additional coverage scenarios for equiv_cov ---------------------------
+
+  Scenario: Reconciliation on empty database returns zeros for equiv_cov
+    When I run the reconciliation job for equiv
+    Then reconciliation returns checked 0, removed 0, kept 0 for equiv_cov
+
+  Scenario: Merge same source and target raises ValidationError for equiv_cov
+    Given a virtual resource exists for equiv
+    Then merging a virtual resource into itself raises ValidationError for equiv_cov
+
+  Scenario: List links on empty database returns empty for equiv_cov
+    When I list equivalence links for equiv
+    Then the list is empty for equiv_cov
+
+  Scenario: Reconciliation removes orphan links for equiv_cov
+    Given a virtual resource exists for equiv
+    And a physical resource exists for equiv
+    And an equivalence link exists between them for equiv
+    And the physical resource is deleted from the database for equiv_cov
+    When I run the reconciliation job for equiv
+    Then the orphan link is removed by reconciliation for equiv_cov
+
+  Scenario: Reconciliation keeps non-content-hash links for equiv_cov
+    Given a virtual resource exists for equiv
+    And a physical resource exists for equiv
+    And a name-type equivalence link exists for equiv_cov
+    When I run the reconciliation job for equiv
+    Then reconciliation reports kept 1 for equiv_cov
+
+  Scenario: Identity key rejects whitespace-only for equiv_cov
+    Then computing an identity key with whitespace-only raises ValidationError for equiv_cov
+
+  Scenario: Merge deduplicates when both virtuals link same physical for equiv_cov
+    Given a virtual resource exists for equiv
+    And a second virtual resource exists for equiv
+    And a physical resource exists for equiv
+    And an equivalence link exists between them for equiv
+    And a link from source virtual to same physical exists for equiv_cov
+    When I merge the source virtual into the target virtual for equiv
+    Then the target has exactly 1 physical link for equiv_cov
+
+  Scenario: Reconcile removes link when content_hash is NULL for equiv_cov
+    Given a virtual resource exists for equiv
+    And a physical resource with no content_hash exists for equiv_cov
+    And a content_hash link from virtual to hashless physical for equiv_cov
+    When I run the reconciliation job for equiv
+    Then reconciliation removed the hashless link for equiv_cov
+
+

features/retry_policy_wiring.feature

@@ -76,7 +76,7 @@ Feature: Retry Policy Wiring for Services
   @registry
   Scenario: Per-service policy defaults with config overrides
     Given I have Settings with retry_service_overrides setting plan_service max_attempts to 10
-    When I create a ServiceRetryWiring from those Settings
+    When I create a ServiceRetryWiring from those default Settings
     Then the plan_service policy should have max_attempts 10
 
   @registry

features/steps/edge_case_plan_steps.py

@@ -755,7 +755,7 @@ def step_try_create_plan_empty_desc(context: Context) -> None:
         context.pydantic_error = exc
 
 
-@then("a Pydantic validation error should be raised")
+@then("a Pydantic validation error should be raised for edge case plan")
 def step_check_pydantic_error(context: Context) -> None:
     assert context.pydantic_error is not None, "Expected a Pydantic validation error"
 

features/steps/plan_executor_coverage_boost_steps.py

@@ -171,7 +171,7 @@ def step_attempt_rollback(context):
     )
 
 
-@then("the rollback result should be False")
+@then("the rollback result should be False for executor")
 def step_verify_rollback_false(context):
     """Verify the rollback result is False."""
     assert context.rollback_result is False

features/steps/plan_explain_steps.py

@@ -330,7 +330,7 @@ def step_json_contains(context: Context, text: str) -> None:
     assert text in context.pe_json_output, f"Expected '{text}' in JSON output"
 
 
-@then("the json output should be valid json")
+@then("the plan explain json output should be valid json")
 def step_json_valid(context: Context) -> None:
     parsed = json.loads(context.pe_json_output)
     assert isinstance(parsed, dict), "Expected a JSON object"

features/steps/project_repository_steps.py

@@ -492,7 +492,7 @@ def step_pr_remove_link_by_id(context: Any, link_id: str) -> None:
     context.pr_remove_result = context.pr_link_repo.remove_link(link_id)
 
 
-@then("the remove result should be False")
+@then("the project remove result should be False")
 def step_pr_remove_false(context: Any) -> None:
     assert context.pr_remove_result is False
 

features/steps/service_retry_wiring_steps.py

@@ -185,7 +185,7 @@ def step_settings_with_overrides(context: Any, value: int) -> None:
         os.environ.pop("CLEVERAGENTS_RETRY_SERVICE_OVERRIDES", None)
 
 
-@when("I create a ServiceRetryWiring from those Settings")
+@when("I create a ServiceRetryWiring from those default Settings")
 def step_create_wiring_from_settings(context: Any) -> None:
     from cleveragents.application.services.service_retry_wiring import (
         ServiceRetryWiring,

features/steps/session_model_steps.py

@@ -446,7 +446,7 @@ def session_model_check_export_messages_count(context: Context) -> None:
 # ---------------------------------------------------------------------------
 
 
-@when("I get the session CLI dict")
+@when("I get the session model CLI dict")
 def session_model_get_cli_dict(context: Context) -> None:
     """Get the CLI dict for the session."""
     context.session_cli_dict = context.session_model.as_cli_dict()

robot/helper_resource_equivalence.py

@@ -0,0 +1,427 @@
+"""Helper script for resource_equivalence.robot smoke tests.
+
+Each subcommand is a self-contained check that prints a sentinel on success.
+"""
+
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+from typing import Any
+
+# Ensure local source tree is importable
+_SRC = str(Path(__file__).resolve().parents[1] / "src")
+if _SRC not in sys.path:
+    sys.path.insert(0, _SRC)
+
+from datetime import UTC, datetime  # noqa: E402
+
+from pydantic import ValidationError  # noqa: E402
+from sqlalchemy import create_engine, event  # noqa: E402
+from sqlalchemy.orm import sessionmaker  # noqa: E402
+from ulid import ULID  # noqa: E402
+
+from cleveragents.domain.models.core.virtual_resource_link import (  # noqa: E402
+    LinkType,
+    VirtualResourceLink,
+)
+from cleveragents.infrastructure.database.models import (  # noqa: E402
+    Base,
+    ResourceModel,
+    ResourceTypeModel,
+)
+
+# ---------------------------------------------------------------------------
+# Shared DB setup
+# ---------------------------------------------------------------------------
+
+
+def _setup_db() -> sessionmaker:
+    """Create in-memory database and return session factory."""
+    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)
+
+
+def _seed_types(session_factory: sessionmaker) -> None:
+    """Register a virtual and a physical resource type."""
+    now_iso = datetime.now(tz=UTC).isoformat()
+    session = session_factory()
+    # Virtual type
+    session.add(
+        ResourceTypeModel(
+            name="file",
+            namespace="builtin",
+            description="Virtual file",
+            resource_kind="virtual",
+            sandbox_strategy="none",
+            user_addable=True,
+            args_schema_json="[]",
+            allowed_parent_types_json="[]",
+            allowed_child_types_json="[]",
+            capabilities_json='{"read":true,"write":false,"sandbox":false,"checkpoint":false}',
+            source="builtin",
+            created_at=now_iso,
+            updated_at=now_iso,
+        )
+    )
+    # Physical type
+    session.add(
+        ResourceTypeModel(
+            name="fs-file",
+            namespace="builtin",
+            description="File system file",
+            resource_kind="physical",
+            sandbox_strategy="none",
+            user_addable=True,
+            args_schema_json="[]",
+            allowed_parent_types_json="[]",
+            allowed_child_types_json="[]",
+            capabilities_json='{"read":true,"write":true,"sandbox":true,"checkpoint":false}',
+            source="builtin",
+            created_at=now_iso,
+            updated_at=now_iso,
+        )
+    )
+    session.commit()
+    session.close()
+
+
+def _make_resource(
+    session_factory: sessionmaker,
+    *,
+    resource_kind: str = "physical",
+    type_name: str = "fs-file",
+    content_hash: str | None = None,
+) -> str:
+    """Create a resource row and return its ULID."""
+    session = session_factory()
+    rid = str(ULID())
+    now_iso = datetime.now(tz=UTC).isoformat()
+    session.add(
+        ResourceModel(
+            resource_id=rid,
+            type_name=type_name,
+            resource_kind=resource_kind,
+            content_hash=content_hash,
+            created_at=now_iso,
+            updated_at=now_iso,
+        )
+    )
+    session.commit()
+    session.close()
+    return rid
+
+
+# ---------------------------------------------------------------------------
+# Subcommands
+# ---------------------------------------------------------------------------
+
+
+def link_type_enum() -> None:
+    """Verify LinkType enum values."""
+    expected = {"content_hash", "name", "identity"}
+    actual = {e.value for e in LinkType}
+    if actual != expected:
+        print(f"FAIL: expected {expected}, got {actual}", file=sys.stderr)
+        sys.exit(1)
+    print("link-type-enum-ok")
+
+
+def domain_model_frozen() -> None:
+    """Verify VirtualResourceLink is frozen."""
+    link = VirtualResourceLink(
+        id=str(ULID()),
+        virtual_resource_id=str(ULID()),
+        physical_resource_id=str(ULID()),
+        equivalence_key="abc123",
+        link_type=LinkType.CONTENT_HASH,
+        confidence=1.0,
+    )
+    try:
+        link.confidence = 0.5  # type: ignore[misc]
+        print("FAIL: should have raised on frozen model", file=sys.stderr)
+        sys.exit(1)
+    except (TypeError, ValueError, ValidationError):
+        pass
+    print("domain-model-frozen-ok")
+
+
+def equivalence_key_computation() -> None:
+    """Verify compute_equivalence_key for content_hash and name."""
+    from cleveragents.application.services.resource_equivalence_service import (
+        compute_equivalence_key,
+    )
+
+    key1 = compute_equivalence_key(content="hello world", link_type="content_hash")
+    if len(key1) != 64:
+        print(
+            f"FAIL: expected 64-char SHA-256 hex, got len={len(key1)}", file=sys.stderr
+        )
+        sys.exit(1)
+
+    key2 = compute_equivalence_key(name="  Main  ", link_type="name")
+    if key2 != "main":
+        print(f"FAIL: expected 'main', got '{key2}'", file=sys.stderr)
+        sys.exit(1)
+
+    key3 = compute_equivalence_key(name="abc123", link_type="identity")
+    if key3 != "abc123":
+        print(f"FAIL: expected 'abc123', got '{key3}'", file=sys.stderr)
+        sys.exit(1)
+
+    print("equivalence-key-computation-ok")
+
+
+def create_and_list_link() -> None:
+    """Verify create_link and list_links round-trip."""
+    from cleveragents.application.services.resource_equivalence_service import (
+        ResourceEquivalenceService,
+    )
+
+    factory = _setup_db()
+    _seed_types(factory)
+
+    virtual_id = _make_resource(factory, resource_kind="virtual", type_name="file")
+    physical_id = _make_resource(
+        factory, resource_kind="physical", type_name="fs-file", content_hash="abc123"
+    )
+
+    svc = ResourceEquivalenceService(factory)
+    link = svc.create_link(virtual_id, physical_id, "abc123", link_type="content_hash")
+
+    if link.virtual_resource_id != virtual_id:
+        print("FAIL: wrong virtual_resource_id", file=sys.stderr)
+        sys.exit(1)
+
+    links = svc.list_links(virtual_resource_id=virtual_id)
+    if len(links) != 1:
+        print(f"FAIL: expected 1 link, got {len(links)}", file=sys.stderr)
+        sys.exit(1)
+
+    print("create-and-list-link-ok")
+
+
+def remove_link() -> None:
+    """Verify remove_link removes an existing link."""
+    from cleveragents.application.services.resource_equivalence_service import (
+        ResourceEquivalenceService,
+    )
+
+    factory = _setup_db()
+    _seed_types(factory)
+
+    virtual_id = _make_resource(factory, resource_kind="virtual", type_name="file")
+    physical_id = _make_resource(
+        factory, resource_kind="physical", type_name="fs-file", content_hash="abc123"
+    )
+
+    svc = ResourceEquivalenceService(factory)
+    svc.create_link(virtual_id, physical_id, "abc123")
+    svc.remove_link(virtual_id, physical_id)
+
+    links = svc.list_links(virtual_resource_id=virtual_id)
+    if len(links) != 0:
+        print(f"FAIL: expected 0 links after remove, got {len(links)}", file=sys.stderr)
+        sys.exit(1)
+
+    print("remove-link-ok")
+
+
+def type_mismatch_guard() -> None:
+    """Verify type mismatch prevents linking incompatible types."""
+    from cleveragents.application.services.resource_equivalence_service import (
+        ResourceEquivalenceService,
+    )
+    from cleveragents.core.exceptions import ValidationError
+
+    factory = _setup_db()
+    _seed_types(factory)
+
+    # Add an incompatible physical type
+    now_iso = datetime.now(tz=UTC).isoformat()
+    session = factory()
+    session.add(
+        ResourceTypeModel(
+            name="git-commit",
+            namespace="builtin",
+            description="Git commit",
+            resource_kind="physical",
+            sandbox_strategy="none",
+            user_addable=True,
+            args_schema_json="[]",
+            allowed_parent_types_json="[]",
+            allowed_child_types_json="[]",
+            capabilities_json='{"read":true,"write":false,"sandbox":false,"checkpoint":false}',
+            source="builtin",
+            created_at=now_iso,
+            updated_at=now_iso,
+        )
+    )
+    session.commit()
+    session.close()
+
+    virtual_id = _make_resource(factory, resource_kind="virtual", type_name="file")
+    physical_id = _make_resource(
+        factory, resource_kind="physical", type_name="git-commit"
+    )
+
+    svc = ResourceEquivalenceService(factory)
+    try:
+        svc.create_link(virtual_id, physical_id, "abc123")
+        print("FAIL: should have raised ValidationError", file=sys.stderr)
+        sys.exit(1)
+    except ValidationError:
+        pass
+
+    print("type-mismatch-guard-ok")
+
+
+def divergence_detection() -> None:
+    """Verify check_divergence removes stale links."""
+    from cleveragents.application.services.resource_equivalence_service import (
+        ResourceEquivalenceService,
+    )
+
+    factory = _setup_db()
+    _seed_types(factory)
+
+    virtual_id = _make_resource(factory, resource_kind="virtual", type_name="file")
+    physical_id = _make_resource(
+        factory, resource_kind="physical", type_name="fs-file", content_hash="hash_v1"
+    )
+
+    svc = ResourceEquivalenceService(factory)
+    svc.create_link(virtual_id, physical_id, "hash_v1")
+
+    # Physical content changed — divergence
+    remaining = svc.check_divergence(physical_id, "hash_v2")
+    if len(remaining) != 0:
+        print(
+            f"FAIL: expected 0 remaining links, got {len(remaining)}", file=sys.stderr
+        )
+        sys.exit(1)
+
+    print("divergence-detection-ok")
+
+
+def reconcile_job() -> None:
+    """Verify reconcile removes stale content_hash links."""
+    from cleveragents.application.services.resource_equivalence_service import (
+        ResourceEquivalenceService,
+    )
+
+    factory = _setup_db()
+    _seed_types(factory)
+
+    virtual_id = _make_resource(factory, resource_kind="virtual", type_name="file")
+    physical_id = _make_resource(
+        factory,
+        resource_kind="physical",
+        type_name="fs-file",
+        content_hash="current_hash",
+    )
+
+    svc = ResourceEquivalenceService(factory)
+    svc.create_link(virtual_id, physical_id, "stale_hash", link_type="content_hash")
+
+    result = svc.reconcile()
+    if result["removed"] != 1:
+        print(f"FAIL: expected 1 removed, got {result['removed']}", file=sys.stderr)
+        sys.exit(1)
+    if result["checked"] != 1:
+        print(f"FAIL: expected 1 checked, got {result['checked']}", file=sys.stderr)
+        sys.exit(1)
+
+    print("reconcile-job-ok")
+
+
+def merge_virtual_resources() -> None:
+    """Verify merge moves links from source to target."""
+    from cleveragents.application.services.resource_equivalence_service import (
+        ResourceEquivalenceService,
+    )
+
+    factory = _setup_db()
+    _seed_types(factory)
+
+    target_id = _make_resource(factory, resource_kind="virtual", type_name="file")
+    source_id = _make_resource(factory, resource_kind="virtual", type_name="file")
+    phys_id = _make_resource(
+        factory, resource_kind="physical", type_name="fs-file", content_hash="hash1"
+    )
+
+    svc = ResourceEquivalenceService(factory)
+    svc.create_link(source_id, phys_id, "hash1")
+
+    merged = svc.merge_virtual_resources(target_id, source_id)
+    if len(merged) != 1:
+        print(f"FAIL: expected 1 merged link, got {len(merged)}", file=sys.stderr)
+        sys.exit(1)
+    if merged[0].virtual_resource_id != target_id:
+        print("FAIL: merged link should point to target", file=sys.stderr)
+        sys.exit(1)
+
+    # Source should have no links
+    source_links = svc.list_links(virtual_resource_id=source_id)
+    if len(source_links) != 0:
+        print(
+            f"FAIL: source should have 0 links, got {len(source_links)}",
+            file=sys.stderr,
+        )
+        sys.exit(1)
+
+    print("merge-virtual-resources-ok")
+
+
+# ---------------------------------------------------------------------------
+# CLI integration tests
+# ---------------------------------------------------------------------------
+
+
+def cli_equivalence_list() -> None:
+    """Test the CLI 'agents resource equivalence list' command."""
+    import subprocess
+
+    result = subprocess.run(
+        [sys.executable, "-m", "cleveragents", "resource", "equivalence", "list"],
+        capture_output=True,
+        text=True,
+        timeout=30,
+    )
+    # The command should succeed (exit 0) even with no links
+    if result.returncode != 0:
+        print(f"FAIL: exit code {result.returncode}", file=sys.stderr)
+        print(f"stderr: {result.stderr}", file=sys.stderr)
+        sys.exit(1)
+    print("cli-equivalence-list-ok")
+
+
+# ---------------------------------------------------------------------------
+# Dispatch
+# ---------------------------------------------------------------------------
+
+_COMMANDS = {
+    "link_type_enum": link_type_enum,
+    "domain_model_frozen": domain_model_frozen,
+    "equivalence_key_computation": equivalence_key_computation,
+    "create_and_list_link": create_and_list_link,
+    "remove_link": remove_link,
+    "type_mismatch_guard": type_mismatch_guard,
+    "divergence_detection": divergence_detection,
+    "reconcile_job": reconcile_job,
+    "merge_virtual_resources": merge_virtual_resources,
+    "cli_equivalence_list": cli_equivalence_list,
+}
+
+if __name__ == "__main__":
+    if len(sys.argv) < 2 or sys.argv[1] not in _COMMANDS:
+        print(f"Usage: {sys.argv[0]} <command>", file=sys.stderr)
+        print(f"Commands: {list(_COMMANDS)}", file=sys.stderr)
+        sys.exit(1)
+    _COMMANDS[sys.argv[1]]()

robot/resource_equivalence.robot

@@ -0,0 +1,101 @@
+*** Settings ***
+Documentation    Smoke tests for Virtual Resource Equivalence Tracking
+...              (domain model, equivalence keys, service CRUD, type guard,
+...              divergence, reconciliation, merge)
+Resource         ${CURDIR}/common.resource
+Suite Setup      Setup Test Environment
+Suite Teardown   Cleanup Test Environment
+
+*** Variables ***
+${HELPER}    ${CURDIR}/helper_resource_equivalence.py
+
+*** Test Cases ***
+LinkType Enum Values
+    [Documentation]    Verify LinkType has content_hash, name, identity
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    link_type_enum
+    ...    cwd=${WORKSPACE}
+    Log    ${result.stdout}
+    Log    ${result.stderr}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    link-type-enum-ok
+
+Domain Model Is Frozen
+    [Documentation]    Verify VirtualResourceLink is immutable
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    domain_model_frozen
+    ...    cwd=${WORKSPACE}
+    Log    ${result.stdout}
+    Log    ${result.stderr}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    domain-model-frozen-ok
+
+Equivalence Key Computation
+    [Documentation]    Verify compute_equivalence_key for all link types
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    equivalence_key_computation
+    ...    cwd=${WORKSPACE}
+    Log    ${result.stdout}
+    Log    ${result.stderr}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    equivalence-key-computation-ok
+
+Create And List Link
+    [Documentation]    Verify create_link and list_links round-trip
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    create_and_list_link
+    ...    cwd=${WORKSPACE}
+    Log    ${result.stdout}
+    Log    ${result.stderr}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    create-and-list-link-ok
+
+Remove Link
+    [Documentation]    Verify remove_link removes an existing link
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    remove_link
+    ...    cwd=${WORKSPACE}
+    Log    ${result.stdout}
+    Log    ${result.stderr}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    remove-link-ok
+
+Type Mismatch Guard
+    [Documentation]    Verify linking incompatible types raises ValidationError
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    type_mismatch_guard
+    ...    cwd=${WORKSPACE}
+    Log    ${result.stdout}
+    Log    ${result.stderr}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    type-mismatch-guard-ok
+
+Divergence Detection
+    [Documentation]    Verify check_divergence removes stale links
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    divergence_detection
+    ...    cwd=${WORKSPACE}
+    Log    ${result.stdout}
+    Log    ${result.stderr}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    divergence-detection-ok
+
+Reconcile Job
+    [Documentation]    Verify reconcile removes stale content_hash links
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    reconcile_job
+    ...    cwd=${WORKSPACE}
+    Log    ${result.stdout}
+    Log    ${result.stderr}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    reconcile-job-ok
+
+Merge Virtual Resources
+    [Documentation]    Verify merge moves links from source to target
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    merge_virtual_resources
+    ...    cwd=${WORKSPACE}
+    Log    ${result.stdout}
+    Log    ${result.stderr}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    merge-virtual-resources-ok
+
+CLI Equivalence List Returns Empty
+    [Documentation]    CLI 'agents resource equivalence list' returns 0 on empty DB
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    cli_equivalence_list
+    ...    cwd=${WORKSPACE}
+    Log    ${result.stdout}
+    Log    ${result.stderr}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    cli-equivalence-list-ok

src/cleveragents/application/container.py

@@ -62,6 +62,9 @@ from cleveragents.application.services.project_service import ProjectService
 from cleveragents.application.services.repo_indexing_service import (
     RepoIndexingService,
 )
+from cleveragents.application.services.resource_equivalence_service import (
+    ResourceEquivalenceService,
+)
 from cleveragents.application.services.resource_file_watcher import (
     ResourceFileWatcher,
 )
@@ -271,6 +274,18 @@ def _build_resource_registry_service(
     return ResourceRegistryService(session_factory=factory)
 
 
+def _build_resource_equivalence_service(
+    database_url: str,
+) -> ResourceEquivalenceService:
+    """Build a ResourceEquivalenceService with a session factory."""
+    from sqlalchemy import create_engine
+    from sqlalchemy.orm import sessionmaker
+
+    engine = create_engine(database_url, echo=False)
+    factory = sessionmaker(bind=engine, expire_on_commit=False)
+    return ResourceEquivalenceService(session_factory=factory)
+
+
 def _build_namespaced_project_repo(
     database_url: str,
 ) -> NamespacedProjectRepository:
@@ -633,6 +648,12 @@ class Container(containers.DeclarativeContainer):
         database_url=database_url,
     )
 
+    # Resource Equivalence Service - virtual-physical link tracking (#334)
+    resource_equivalence_service = providers.Factory(
+        _build_resource_equivalence_service,
+        database_url=database_url,
+    )
+
     # Namespaced Project Repository
     namespaced_project_repo = providers.Factory(
         _build_namespaced_project_repo,

src/cleveragents/application/services/__init__.py

@@ -559,6 +559,14 @@ _LAZY_IMPORTS: dict[str, tuple[str, str]] = {
     "ValidationPipeline": ("validation_pipeline", "ValidationPipeline"),
     "ValidationResult": ("validation_pipeline", "ValidationResult"),
     "ValidationSummary": ("validation_pipeline", "ValidationSummary"),
+    "ResourceEquivalenceService": (
+        "resource_equivalence_service",
+        "ResourceEquivalenceService",
+    ),
+    "compute_equivalence_key": (
+        "resource_equivalence_service",
+        "compute_equivalence_key",
+    ),
 }
 
 

src/cleveragents/application/services/_resource_equivalence_helpers.py

@@ -0,0 +1,228 @@
+"""Internal helpers for resource equivalence tracking.
+
+Extracted from ``resource_equivalence_service.py`` to keep that module
+within the 500-line limit.  Contains:
+
+- ``compute_equivalence_key`` — public helper for computing hash/name/identity keys.
+- ``ALLOWED_PAIRINGS`` — virtual ↔ physical type compatibility map.
+- ``db_link_to_domain`` — ORM row → domain model converter.
+- ``validate_link_inputs`` — shared input validation for link creation.
+
+Based on:
+
+- Specification: Equivalence linking (lines 24397-24417)
+- Issue #334: Virtual resource equivalence tracking
+"""
+
+from __future__ import annotations
+
+import hashlib
+from datetime import datetime
+from pathlib import Path
+
+from cleveragents.core.exceptions import ValidationError
+from cleveragents.domain.models.core.virtual_resource_link import (
+    LinkType,
+    VirtualResourceLink,
+)
+from cleveragents.infrastructure.database.models import (
+    VirtualResourceLinkModel,
+)
+
+# ---------------------------------------------------------------------------
+# Equivalence key computation
+# ---------------------------------------------------------------------------
+
+
+def compute_equivalence_key(
+    content: str | None = None,
+    name: str | None = None,
+    *,
+    link_type: str = "content_hash",
+) -> str:
+    """Compute an equivalence key for auto-linking during discovery.
+
+    For ``content_hash`` link type, computes SHA-256 of the content.
+    For ``name`` link type, normalises the name (lowercase, stripped).
+    For ``identity`` link type, uses the name directly as the identity.
+
+    Args:
+        content: Content bytes/string for hash-based equivalence.
+        name: Name or identity string for name/identity equivalence.
+        link_type: One of ``content_hash``, ``name``, or ``identity``.
+
+    Returns:
+        The computed equivalence key string.
+
+    Raises:
+        ValidationError: If required input is missing for the link type.
+    """
+    if link_type == "content_hash":
+        if content is None:
+            raise ValidationError(
+                message="Content is required for content_hash equivalence",
+                details={"link_type": link_type},
+            )
+        return hashlib.sha256(content.encode("utf-8")).hexdigest()
+    elif link_type == "name":
+        if name is None:
+            raise ValidationError(
+                message="Name is required for name equivalence",
+                details={"link_type": link_type},
+            )
+        stripped = name.strip()
+        if not stripped:
+            raise ValidationError(
+                message="Name cannot be empty or whitespace-only",
+                details={"link_type": link_type},
+            )
+        return stripped.lower()
+    elif link_type == "identity":
+        if name is None:
+            raise ValidationError(
+                message="Identity value is required for identity equivalence",
+                details={"link_type": link_type},
+            )
+        # Identity keys are opaque pass-through values (e.g., commit
+        # hashes, UUIDs).  Only reject truly empty strings; do NOT
+        # strip whitespace as that would alter the identity value.
+        if not name or not name.strip():
+            raise ValidationError(
+                message="Identity value cannot be empty or whitespace-only",
+                details={"link_type": link_type},
+            )
+        return name
+    else:
+        raise ValidationError(
+            message=f"Unknown link type: {link_type}",
+            details={"link_type": link_type},
+        )
+
+
+def compute_equivalence_key_from_path(
+    path: Path,
+    *,
+    chunk_size: int = 65536,
+) -> str:
+    """Stream-hash a file to compute a content_hash equivalence key.
+
+    Reads the file in chunks so multi-GB files do not cause OOM.
+
+    Args:
+        path: Path to the file to hash.
+        chunk_size: Read buffer size in bytes (default 64 KiB).
+
+    Returns:
+        SHA-256 hex digest of the file content.
+
+    Raises:
+        ValidationError: If the file does not exist or is not readable.
+    """
+    if not path.is_file():
+        raise ValidationError(
+            message=f"Path is not a readable file: {path}",
+            details={"path": str(path)},
+        )
+    hasher = hashlib.sha256()
+    with path.open("rb") as fh:
+        while True:
+            chunk = fh.read(chunk_size)
+            if not chunk:
+                break
+            hasher.update(chunk)
+    return hasher.hexdigest()
+
+
+# ---------------------------------------------------------------------------
+# Allowed type pairings for virtual-physical linking
+# ---------------------------------------------------------------------------
+
+# Maps virtual type name to allowed physical type names.
+# If a virtual type is not in this table, any physical type is allowed
+# (relaxed mode for user-defined types).
+#
+# Virtual type names are unprefixed per spec (line 10223):
+# "file", "directory", "commit", etc. — NOT "v-file", "v-directory".
+ALLOWED_PAIRINGS: dict[str, tuple[str, ...]] = {
+    "file": ("fs-file", "git-tree-entry"),
+    "directory": ("fs-directory", "git-tree"),
+    "commit": ("git-commit",),
+    "branch": ("git-branch",),
+    "tag": ("git-tag",),
+    "remote": ("git-remote",),
+    "submodule": ("git-submodule",),
+    "tree": ("git-tree",),
+    "symlink": ("fs-symlink", "git-tree-entry"),
+}
+
+
+# ---------------------------------------------------------------------------
+# ORM → domain conversion
+# ---------------------------------------------------------------------------
+
+
+def db_link_to_domain(row: VirtualResourceLinkModel) -> VirtualResourceLink:
+    """Convert a VirtualResourceLinkModel DB row to domain object."""
+    return VirtualResourceLink(
+        id=str(row.id),
+        virtual_resource_id=str(row.virtual_resource_id),
+        physical_resource_id=str(row.physical_resource_id),
+        equivalence_key=str(row.equivalence_key),
+        link_type=LinkType(str(row.link_type)),
+        confidence=float(str(row.confidence)),
+        created_at=datetime.fromisoformat(str(row.created_at)),
+        updated_at=datetime.fromisoformat(str(row.updated_at)),
+    )
+
+
+# ---------------------------------------------------------------------------
+# Input validation
+# ---------------------------------------------------------------------------
+
+
+def validate_link_inputs(
+    equivalence_key: str,
+    confidence: float,
+    link_type: str,
+) -> str:
+    """Validate and normalise inputs for link creation.
+
+    Args:
+        equivalence_key: The computed equivalence key.
+        confidence: Confidence score (0.0-1.0).
+        link_type: How equivalence was established.
+
+    Returns:
+        The stripped equivalence key.
+
+    Raises:
+        ValidationError: If any input is invalid.
+    """
+    if not equivalence_key or not equivalence_key.strip():
+        raise ValidationError(
+            message="equivalence_key cannot be empty or whitespace-only",
+            details={"equivalence_key": equivalence_key},
+        )
+    if confidence < 0.0 or confidence > 1.0:
+        raise ValidationError(
+            message="confidence must be between 0.0 and 1.0",
+            details={"confidence": confidence},
+        )
+    if link_type not in ("content_hash", "name", "identity"):
+        raise ValidationError(
+            message=f"Invalid link_type: {link_type}",
+            details={"link_type": link_type},
+        )
+    # Identity keys are opaque pass-through; do NOT strip.
+    if link_type == "identity":
+        return equivalence_key
+    return equivalence_key.strip()
+
+
+__all__ = [
+    "ALLOWED_PAIRINGS",
+    "compute_equivalence_key",
+    "compute_equivalence_key_from_path",
+    "db_link_to_domain",
+    "validate_link_inputs",
+]

src/cleveragents/application/services/_resource_equivalence_reconciler.py

@@ -0,0 +1,128 @@
+"""Batch reconciliation for virtual resource equivalence links.
+
+Extracted from ``resource_equivalence_service.py`` to keep that module
+within the 500-line limit.
+
+Issue #334 · spec section "Virtual Resource Equivalence".
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+import structlog
+
+from cleveragents.infrastructure.database.models import (
+    ResourceModel,
+    VirtualResourceLinkModel,
+)
+
+if TYPE_CHECKING:
+    from sqlalchemy.orm import Session
+
+logger = structlog.get_logger(__name__)
+
+#: SQLite SQLITE_MAX_VARIABLE_NUMBER safe limit for IN clauses.
+_IN_CHUNK_SIZE = 500
+
+
+def reconcile_links(session: Session) -> dict[str, int]:
+    """Batch-verify content_hash links; remove stale ones.
+
+    For each ``content_hash`` link, the physical resource's current
+    ``content_hash`` is compared to the link's ``equivalence_key``.
+    Links are removed when:
+
+    - The physical resource no longer exists (orphan).
+    - The physical resource has no ``content_hash`` (cannot verify).
+    - The hashes do not match (content diverged).
+
+    Non-``content_hash`` links (``name``, ``identity``) are kept
+    unconditionally.
+
+    Args:
+        session: Active SQLAlchemy session (caller manages commit/rollback).
+
+    Returns:
+        Dict with ``checked``, ``removed``, ``kept`` counts.
+    """
+    # Count non-content_hash links (kept unconditionally)
+    kept = (
+        session.query(VirtualResourceLinkModel)
+        .filter(VirtualResourceLinkModel.link_type != "content_hash")
+        .count()
+    )
+
+    # Load only content_hash links (server-side filter)
+    hash_links = (
+        session.query(VirtualResourceLinkModel)
+        .filter(VirtualResourceLinkModel.link_type == "content_hash")
+        .all()
+    )
+
+    checked = len(hash_links)
+    removed = 0
+
+    if hash_links:
+        # Batch-load physical resources in chunks to avoid
+        # exceeding SQLite's SQLITE_MAX_VARIABLE_NUMBER limit.
+        physical_ids = list({str(link.physical_resource_id) for link in hash_links})
+        hash_by_id: dict[str, str | None] = {}
+        for i in range(0, len(physical_ids), _IN_CHUNK_SIZE):
+            chunk = physical_ids[i : i + _IN_CHUNK_SIZE]
+            physical_rows = (
+                session.query(ResourceModel)
+                .filter(ResourceModel.resource_id.in_(chunk))
+                .all()
+            )
+            for row in physical_rows:
+                hash_by_id[str(row.resource_id)] = (
+                    str(row.content_hash) if row.content_hash is not None else None
+                )
+
+        for link in hash_links:
+            phys_id = str(link.physical_resource_id)
+            if phys_id not in hash_by_id:
+                # Physical resource no longer exists — orphan link
+                session.delete(link)
+                removed += 1
+                logger.warning(
+                    "equivalence_link.orphan_removed",
+                    link_id=str(link.id),
+                    physical_resource_id=phys_id,
+                )
+            elif hash_by_id[phys_id] is None:
+                # Physical resource has no content_hash — cannot verify
+                session.delete(link)
+                removed += 1
+                logger.warning(
+                    "equivalence_link.no_hash_removed",
+                    link_id=str(link.id),
+                    physical_resource_id=phys_id,
+                )
+            elif hash_by_id[phys_id] != str(link.equivalence_key):
+                # Hash mismatch — content diverged
+                session.delete(link)
+                removed += 1
+                logger.info(
+                    "equivalence_link.hash_mismatch_removed",
+                    link_id=str(link.id),
+                    stored_key=str(link.equivalence_key)[:8],
+                    current_hash=(hash_by_id[phys_id] or "")[:8],
+                )
+            else:
+                kept += 1
+
+    session.flush()
+
+    logger.info(
+        "equivalence_link.reconciliation_complete",
+        checked=checked,
+        removed=removed,
+        kept=kept,
+    )
+
+    return {"checked": checked, "removed": removed, "kept": kept}
+
+
+__all__ = ["reconcile_links"]

src/cleveragents/application/services/resource_equivalence_service.py

@@ -0,0 +1,498 @@
+"""Resource Equivalence Service — CRUD, merge, divergence, and reconciliation.
+
+Manages virtual-to-physical resource equivalence links.  Physical resources
+take precedence; divergent links are removed automatically.
+
+See also: ``_resource_equivalence_helpers.py`` for ``compute_equivalence_key``,
+type-pairing map, and shared validators.
+
+Issue #334 · spec section "Virtual Resource Equivalence".
+"""
+
+from __future__ import annotations
+
+from collections.abc import Callable
+from datetime import UTC, datetime
+from typing import TYPE_CHECKING
+
+import structlog
+from sqlalchemy.exc import IntegrityError
+from ulid import ULID
+
+from cleveragents.application.services._resource_equivalence_helpers import (
+    ALLOWED_PAIRINGS,
+    compute_equivalence_key,
+    db_link_to_domain,
+    validate_link_inputs,
+)
+from cleveragents.core.exceptions import NotFoundError, ValidationError
+from cleveragents.domain.models.core.virtual_resource_link import (
+    LinkType,
+    VirtualResourceLink,
+)
+from cleveragents.infrastructure.database.models import (
+    ResourceModel,
+    VirtualResourceLinkModel,
+)
+
+if TYPE_CHECKING:
+    from sqlalchemy.orm import Session
+
+logger = structlog.get_logger(__name__)
+
+
+class ResourceEquivalenceService:
+    """Virtual resource equivalence tracking (session-factory pattern).
+
+    INVARIANTS (must hold after every public method returns):
+    1. Every link references valid resource IDs in the resources table.
+    2. virtual_resource_id → resource_kind='virtual'.
+    3. physical_resource_id → resource_kind='physical'.
+    4. (virtual_resource_id, physical_resource_id) is unique.
+    5. confidence in [0.0, 1.0]; link_type in {content_hash, name, identity}.
+    6. equivalence_key is never empty or whitespace-only.
+
+    Args:
+        session_factory: Callable returning a new SQLAlchemy ``Session``.
+    """
+
+    def __init__(self, session_factory: Callable[[], Session]) -> None:
+        self._session_factory = session_factory
+
+    def _session(self) -> Session:
+        return self._session_factory()
+
+    @staticmethod
+    def _verify_resource(
+        session: Session,
+        resource_id: str,
+        expected_kind: str,
+    ) -> ResourceModel:
+        """Load a resource row, raising if missing or wrong kind."""
+        row = session.query(ResourceModel).filter_by(resource_id=resource_id).first()
+        if row is None:
+            raise NotFoundError(resource_type="resource", resource_id=resource_id)
+        if str(row.resource_kind) != expected_kind:
+            raise ValidationError(
+                message=(
+                    f"Resource '{resource_id}' is not {expected_kind} "
+                    f"(kind={row.resource_kind})"
+                ),
+                details={
+                    "resource_id": resource_id,
+                    "resource_kind": str(row.resource_kind),
+                },
+            )
+        return row
+
+    # -- Create Link ---------------------------------------------------------
+
+    def create_link(
+        self,
+        virtual_resource_id: str,
+        physical_resource_id: str,
+        equivalence_key: str,
+        *,
+        link_type: str = LinkType.CONTENT_HASH,
+        confidence: float = 1.0,
+    ) -> VirtualResourceLink:
+        """Create a virtual-to-physical equivalence link.
+
+        Args:
+            virtual_resource_id: ULID of the virtual resource.
+            physical_resource_id: ULID of the physical resource.
+            equivalence_key: Computed equivalence key.
+            link_type: How equivalence was established.
+            confidence: Confidence score (0.0-1.0).
+
+        Returns:
+            The created ``VirtualResourceLink`` domain object.
+
+        Raises:
+            NotFoundError: If either resource does not exist.
+            ValidationError: On bad inputs, type mismatch, or duplicate link.
+        """
+        equivalence_key = validate_link_inputs(equivalence_key, confidence, link_type)
+
+        session = self._session()
+        try:
+            virtual_row = self._verify_resource(session, virtual_resource_id, "virtual")
+            physical_row = self._verify_resource(
+                session,
+                physical_resource_id,
+                "physical",
+            )
+
+            # Guard: prevent linking incompatible types
+            virtual_type = str(virtual_row.type_name)
+            physical_type = str(physical_row.type_name)
+            if virtual_type in ALLOWED_PAIRINGS:
+                allowed = ALLOWED_PAIRINGS[virtual_type]
+                if physical_type not in allowed:
+                    raise ValidationError(
+                        message=(
+                            f"Type mismatch: virtual type '{virtual_type}' "
+                            f"cannot link to physical type '{physical_type}'. "
+                            f"Allowed: {', '.join(allowed)}"
+                        ),
+                        details={
+                            "virtual_type": virtual_type,
+                            "physical_type": physical_type,
+                            "allowed_physical_types": list(allowed),
+                        },
+                    )
+
+            # Check for duplicate link
+            existing = (
+                session.query(VirtualResourceLinkModel)
+                .filter_by(
+                    virtual_resource_id=virtual_resource_id,
+                    physical_resource_id=physical_resource_id,
+                )
+                .first()
+            )
+            if existing is not None:
+                raise ValidationError(
+                    message=(
+                        f"Link already exists: "
+                        f"{virtual_resource_id} -> {physical_resource_id}"
+                    ),
+                    details={
+                        "virtual_resource_id": virtual_resource_id,
+                        "physical_resource_id": physical_resource_id,
+                    },
+                )
+
+            link_id = str(ULID())
+            now_iso = datetime.now(tz=UTC).isoformat()
+
+            db_link = VirtualResourceLinkModel(
+                id=link_id,
+                virtual_resource_id=virtual_resource_id,
+                physical_resource_id=physical_resource_id,
+                equivalence_key=equivalence_key,
+                link_type=link_type,
+                confidence=confidence,
+                created_at=now_iso,
+                updated_at=now_iso,
+            )
+            session.add(db_link)
+            try:
+                session.flush()
+                session.commit()
+            except IntegrityError as exc:
+                session.rollback()
+                raise ValidationError(
+                    message=(
+                        f"Link already exists: "
+                        f"{virtual_resource_id} -> {physical_resource_id}"
+                    ),
+                    details={
+                        "virtual_resource_id": virtual_resource_id,
+                        "physical_resource_id": physical_resource_id,
+                    },
+                ) from exc
+
+            logger.info(
+                "equivalence_link.created",
+                link_id=link_id,
+                virtual_resource_id=virtual_resource_id,
+                physical_resource_id=physical_resource_id,
+                equivalence_key=equivalence_key[:8] + "...",
+                link_type=link_type,
+            )
+
+            return VirtualResourceLink(
+                id=link_id,
+                virtual_resource_id=virtual_resource_id,
+                physical_resource_id=physical_resource_id,
+                equivalence_key=equivalence_key,
+                link_type=LinkType(link_type),
+                confidence=confidence,
+                created_at=datetime.fromisoformat(now_iso),
+                updated_at=datetime.fromisoformat(now_iso),
+            )
+        except Exception:
+            session.rollback()
+            raise
+        finally:
+            session.close()
+
+    # -- List Links ----------------------------------------------------------
+
+    def list_links(
+        self,
+        virtual_resource_id: str | None = None,
+        physical_resource_id: str | None = None,
+        *,
+        limit: int = 200,
+        offset: int = 0,
+    ) -> list[VirtualResourceLink]:
+        """List equivalence links, optionally filtered by virtual/physical ID.
+
+        Args:
+            virtual_resource_id: Filter by virtual resource ULID.
+            physical_resource_id: Filter by physical resource ULID.
+            limit: Maximum number of results (default 200).
+            offset: Number of results to skip (default 0).
+
+        Returns:
+            List of ``VirtualResourceLink`` domain objects ordered by created_at.
+        """
+        session = self._session()
+        try:
+            query = session.query(VirtualResourceLinkModel)
+            if virtual_resource_id is not None:
+                query = query.filter_by(virtual_resource_id=virtual_resource_id)
+            if physical_resource_id is not None:
+                query = query.filter_by(physical_resource_id=physical_resource_id)
+            query = query.order_by(VirtualResourceLinkModel.created_at)
+            query = query.limit(limit).offset(offset)
+            rows = query.all()
+            return [db_link_to_domain(row) for row in rows]
+        finally:
+            session.close()
+
+    # -- Remove Link ---------------------------------------------------------
+
+    def remove_link(
+        self,
+        virtual_resource_id: str,
+        physical_resource_id: str,
+    ) -> None:
+        """Remove an equivalence link.
+
+        Args:
+            virtual_resource_id: ULID of the virtual resource.
+            physical_resource_id: ULID of the physical resource.
+
+        Raises:
+            NotFoundError: If the link does not exist.
+        """
+        session = self._session()
+        try:
+            row = (
+                session.query(VirtualResourceLinkModel)
+                .filter_by(
+                    virtual_resource_id=virtual_resource_id,
+                    physical_resource_id=physical_resource_id,
+                )
+                .first()
+            )
+            if row is None:
+                raise NotFoundError(
+                    resource_type="virtual_resource_link",
+                    resource_id=(f"{virtual_resource_id} -> {physical_resource_id}"),
+                )
+            session.delete(row)
+            session.flush()
+            session.commit()
+
+            logger.info(
+                "equivalence_link.removed",
+                virtual_resource_id=virtual_resource_id,
+                physical_resource_id=physical_resource_id,
+            )
+        except Exception:
+            session.rollback()
+            raise
+        finally:
+            session.close()
+
+    # -- Merge Virtual Resources ---------------------------------------------
+
+    def merge_virtual_resources(
+        self,
+        target_virtual_id: str,
+        source_virtual_id: str,
+    ) -> list[VirtualResourceLink]:
+        """Move all links from *source* virtual resource to *target*.
+
+        Duplicate links (physical already linked to target) are skipped.
+        Returns all links for the target after the merge.
+
+        Raises:
+            ValidationError: If source equals target or concurrent conflict.
+            NotFoundError: If either virtual resource does not exist.
+        """
+        if target_virtual_id == source_virtual_id:
+            raise ValidationError(
+                message="Cannot merge a virtual resource into itself",
+                details={
+                    "target_virtual_id": target_virtual_id,
+                    "source_virtual_id": source_virtual_id,
+                },
+            )
+
+        session = self._session()
+        try:
+            for res_id in (target_virtual_id, source_virtual_id):
+                self._verify_resource(session, res_id, "virtual")
+
+            # Get all links from source
+            source_links = (
+                session.query(VirtualResourceLinkModel)
+                .filter_by(virtual_resource_id=source_virtual_id)
+                .all()
+            )
+
+            # Batch-load existing target links to avoid N+1 queries
+            target_links = (
+                session.query(VirtualResourceLinkModel)
+                .filter_by(virtual_resource_id=target_virtual_id)
+                .all()
+            )
+            target_physical_ids = {
+                str(lnk.physical_resource_id) for lnk in target_links
+            }
+
+            now_iso = datetime.now(tz=UTC).isoformat()
+            moved_count = 0
+
+            for link in source_links:
+                physical_id = str(link.physical_resource_id)
+                if physical_id not in target_physical_ids:
+                    # Move the link to the target (setattr for ORM Column compat)
+                    setattr(link, "virtual_resource_id", target_virtual_id)  # noqa: B010
+                    setattr(link, "updated_at", now_iso)  # noqa: B010
+                    target_physical_ids.add(physical_id)
+                    moved_count += 1
+                else:
+                    # Skip duplicate — delete the source link
+                    session.delete(link)
+
+            try:
+                session.flush()
+            except IntegrityError as exc:
+                session.rollback()
+                raise ValidationError(
+                    message=(
+                        "Merge conflict: concurrent modification detected. "
+                        "Retry the merge operation."
+                    ),
+                    details={
+                        "target_virtual_id": target_virtual_id,
+                        "source_virtual_id": source_virtual_id,
+                    },
+                ) from exc
+
+            # Query result within the SAME session to avoid dual-session issues
+            result_rows = (
+                session.query(VirtualResourceLinkModel)
+                .filter_by(virtual_resource_id=target_virtual_id)
+                .order_by(VirtualResourceLinkModel.created_at)
+                .all()
+            )
+            result_links = [db_link_to_domain(r) for r in result_rows]
+
+            session.commit()
+
+            logger.info(
+                "equivalence_link.merged",
+                source_virtual_id=source_virtual_id,
+                target_virtual_id=target_virtual_id,
+                moved_count=moved_count,
+            )
+
+            return result_links
+        except Exception:
+            session.rollback()
+            raise
+        finally:
+            session.close()
+
+    # -- Divergence Detection ------------------------------------------------
+
+    def check_divergence(
+        self,
+        physical_resource_id: str,
+        new_equivalence_key: str,
+    ) -> list[VirtualResourceLink]:
+        """Remove *content_hash* links whose key no longer matches.
+
+        Only ``content_hash`` links are compared; ``name``/``identity``
+        links are kept unconditionally.  Returns surviving links.
+
+        Raises:
+            ValidationError: If ``new_equivalence_key`` is empty.
+        """
+        if not new_equivalence_key or not new_equivalence_key.strip():
+            raise ValidationError(
+                message="new_equivalence_key cannot be empty",
+                details={"new_equivalence_key": new_equivalence_key},
+            )
+
+        new_equivalence_key = new_equivalence_key.strip()
+        session = self._session()
+        try:
+            links = (
+                session.query(VirtualResourceLinkModel)
+                .filter_by(physical_resource_id=physical_resource_id)
+                .all()
+            )
+
+            remaining: list[VirtualResourceLink] = []
+            now_iso = datetime.now(tz=UTC).isoformat()
+
+            for link in links:
+                link_type = str(link.link_type)
+
+                if link_type != "content_hash":
+                    # Non-content-hash links are kept unconditionally
+                    remaining.append(db_link_to_domain(link))
+                    continue
+
+                if str(link.equivalence_key) != new_equivalence_key:
+                    logger.warning(
+                        "equivalence_link.divergence_detected",
+                        physical_resource_id=physical_resource_id,
+                        virtual_resource_id=str(link.virtual_resource_id),
+                        old_key=str(link.equivalence_key)[:8],
+                        new_key=new_equivalence_key[:8],
+                    )
+                    session.delete(link)
+                else:
+                    setattr(link, "updated_at", now_iso)  # noqa: B010
+                    remaining.append(db_link_to_domain(link))
+
+            session.flush()
+            session.commit()
+
+            return remaining
+        except Exception:
+            session.rollback()
+            raise
+        finally:
+            session.close()
+
+    # -- Reconciliation Job --------------------------------------------------
+
+    def reconcile(self) -> dict[str, int]:
+        """Batch-verify content_hash links; remove stale ones.
+
+        Delegates to ``_resource_equivalence_reconciler.reconcile_links``
+        (extracted for the 500-line module limit).
+
+        Returns:
+            Dict with ``checked``, ``removed``, ``kept`` counts.
+        """
+        from cleveragents.application.services._resource_equivalence_reconciler import (
+            reconcile_links,
+        )
+
+        session = self._session()
+        try:
+            result = reconcile_links(session)
+            session.commit()
+            return result
+        except Exception:
+            session.rollback()
+            raise
+        finally:
+            session.close()
+
+
+__all__ = [
+    "ResourceEquivalenceService",
+    "compute_equivalence_key",
+]

src/cleveragents/cli/commands/resource.py

@@ -67,6 +67,9 @@ from cleveragents.application.container import get_container
 from cleveragents.application.services.resource_registry_service import (
     ResourceRegistryService,
 )
+from cleveragents.cli.commands.resource_equivalence import (
+    app as equivalence_app,
+)
 from cleveragents.cli.formatting import OutputFormat, format_output
 from cleveragents.cli.renderers import _get_console
 from cleveragents.core.exceptions import (
@@ -97,6 +100,9 @@ type_app = typer.Typer(
 )
 app.add_typer(type_app, name="type")
 
+# Sub-app for ``agents resource equivalence ...``
+app.add_typer(equivalence_app, name="equivalence")
+
 console = _get_console()
 
 _FORMAT_HELP = "Output format: json, yaml, plain, table, or rich (default: rich)"

src/cleveragents/cli/commands/resource_equivalence.py

@@ -0,0 +1,307 @@
+"""Resource equivalence CLI commands for CleverAgents.
+
+The ``agents resource equivalence`` command group manages virtual-to-physical
+resource equivalence links.
+
+## Commands
+
+| Command                              | Description                          |
+|--------------------------------------|--------------------------------------|
+| ``agents resource equivalence list`` | List equivalence links               |
+| ``agents resource equivalence add``  | Add an equivalence link              |
+| ``agents resource equivalence remove``| Remove an equivalence link           |
+
+## Example Usage
+
+```bash
+# List all links for a virtual resource
+agents resource equivalence list --virtual 01HXYZ...
+
+# Add an equivalence link
+agents resource equivalence add VIRT_ULID PHYS_ULID --key abc123
+
+# Remove an equivalence link
+agents resource equivalence remove VIRT_ULID PHYS_ULID
+```
+
+Based on Issue #334: Virtual resource equivalence tracking.
+"""
+
+from __future__ import annotations
+
+from typing import Annotated
+
+import typer
+from rich.table import Table
+
+from cleveragents.application.container import get_container
+from cleveragents.application.services.resource_equivalence_service import (
+    ResourceEquivalenceService,
+)
+from cleveragents.cli.formatting import OutputFormat, format_output
+from cleveragents.cli.renderers import _get_console
+from cleveragents.core.exceptions import (
+    CleverAgentsError,
+    NotFoundError,
+    ValidationError,
+)
+from cleveragents.domain.models.core.virtual_resource_link import (
+    VirtualResourceLink,
+)
+
+# Sub-app for ``agents resource equivalence ...``
+app = typer.Typer(
+    help="Manage virtual resource equivalence sets.",
+)
+
+console = _get_console()
+
+_FORMAT_HELP = "Output format: json, yaml, plain, table, or rich (default: rich)"
+
+
+def _validate_ulid(value: str, name: str) -> None:
+    """Validate that a string looks like a ULID (26 uppercase alphanumeric)."""
+    if len(value) != 26 or not value.isalnum():
+        console.print(
+            f"[red]Invalid ULID for {name}:[/red] '{value}'. "
+            f"Expected 26 alphanumeric characters."
+        )
+        raise typer.Abort()
+
+
+def _get_equivalence_service() -> ResourceEquivalenceService:
+    """Get the ResourceEquivalenceService from the DI container."""
+    container = get_container()
+    service: ResourceEquivalenceService = container.resource_equivalence_service()
+    return service
+
+
+def _equiv_link_dict(link: VirtualResourceLink) -> dict[str, object]:
+    """Convert a VirtualResourceLink to a plain dict for output."""
+    return {
+        "id": link.id,
+        "virtual_resource_id": link.virtual_resource_id,
+        "physical_resource_id": link.physical_resource_id,
+        "equivalence_key": link.equivalence_key,
+        "link_type": str(link.link_type),
+        "confidence": link.confidence,
+        "created_at": link.created_at.isoformat(),
+        "updated_at": link.updated_at.isoformat(),
+    }
+
+
+@app.command("list")
+def equivalence_list(
+    virtual_id: Annotated[
+        str | None,
+        typer.Option("--virtual", "-v", help="Filter by virtual resource ULID"),
+    ] = None,
+    physical_id: Annotated[
+        str | None,
+        typer.Option("--physical", "-p", help="Filter by physical resource ULID"),
+    ] = None,
+    fmt: Annotated[
+        str,
+        typer.Option("--format", "-f", help=_FORMAT_HELP),
+    ] = "rich",
+) -> None:
+    """List virtual resource equivalence links.
+
+    Examples:
+        agents resource equivalence list
+        agents resource equivalence list --virtual 01HXYZ...
+        agents resource equivalence list --physical 01HABC...
+        agents resource equivalence list --format json
+    """
+    try:
+        service = _get_equivalence_service()
+        links = service.list_links(
+            virtual_resource_id=virtual_id,
+            physical_resource_id=physical_id,
+        )
+
+        if not links:
+            console.print("[yellow]No equivalence links found.[/yellow]")
+            return
+
+        if fmt != OutputFormat.RICH.value:
+            data = [_equiv_link_dict(lnk) for lnk in links]
+            console.print(format_output(data, fmt))
+            return
+
+        table = Table(title=f"Equivalence Links ({len(links)} total)")
+        table.add_column("ID", style="dim")
+        table.add_column("Virtual", style="cyan")
+        table.add_column("Physical", style="blue")
+        table.add_column("Key", style="magenta")
+        table.add_column("Type", style="green")
+        table.add_column("Confidence")
+
+        for lnk in links:
+            lid = lnk.id
+            if len(lid) > 12:
+                lid = lid[:12] + "..."
+            vid = lnk.virtual_resource_id
+            if len(vid) > 12:
+                vid = vid[:12] + "..."
+            pid = lnk.physical_resource_id
+            if len(pid) > 12:
+                pid = pid[:12] + "..."
+            key = lnk.equivalence_key
+            if len(key) > 20:
+                key = key[:17] + "..."
+            table.add_row(
+                lid,
+                vid,
+                pid,
+                key,
+                str(lnk.link_type),
+                f"{lnk.confidence:.2f}",
+            )
+
+        console.print(table)
+
+    except CleverAgentsError as exc:
+        console.print(f"[red]Error:[/red] {exc.message}")
+        raise typer.Abort() from exc
+    except Exception as exc:
+        if isinstance(exc, (typer.Abort, typer.Exit)):
+            raise
+        console.print(f"[red]Unexpected error:[/red] {exc}")
+        raise typer.Abort() from exc
+
+
+@app.command("add")
+def equivalence_add(
+    virtual_id: Annotated[
+        str,
+        typer.Argument(help="Virtual resource ULID"),
+    ],
+    physical_id: Annotated[
+        str,
+        typer.Argument(help="Physical resource ULID"),
+    ],
+    equivalence_key: Annotated[
+        str,
+        typer.Option("--key", "-k", help="Equivalence key (hash, name, or identity)"),
+    ],
+    link_type: Annotated[
+        str,
+        typer.Option("--type", "-t", help="Link type: content_hash, name, or identity"),
+    ] = "content_hash",
+    confidence: Annotated[
+        float,
+        typer.Option("--confidence", "-c", help="Confidence score (0.0-1.0)"),
+    ] = 1.0,
+    fmt: Annotated[
+        str,
+        typer.Option("--format", "-f", help=_FORMAT_HELP),
+    ] = "rich",
+) -> None:
+    """Add a virtual resource equivalence link.
+
+    Examples:
+        agents resource equivalence add VIRT_ULID PHYS_ULID --key abc123
+        agents resource equivalence add VIRT_ULID PHYS_ULID --key main --type name
+    """
+    _validate_ulid(virtual_id, "virtual_id")
+    _validate_ulid(physical_id, "physical_id")
+    try:
+        service = _get_equivalence_service()
+        link = service.create_link(
+            virtual_resource_id=virtual_id,
+            physical_resource_id=physical_id,
+            equivalence_key=equivalence_key,
+            link_type=link_type,
+            confidence=confidence,
+        )
+
+        if fmt != OutputFormat.RICH.value:
+            data = _equiv_link_dict(link)
+            console.print(format_output(data, fmt))
+            return
+
+        console.print(
+            f"[green]Added equivalence link:[/green] "
+            f"{virtual_id} -> {physical_id} (key={equivalence_key})"
+        )
+
+    except NotFoundError as exc:
+        console.print(f"[red]Resource not found:[/red] {exc.resource_id}")
+        raise typer.Abort() from exc
+    except ValidationError as exc:
+        console.print(f"[red]Validation error:[/red] {exc.message}")
+        raise typer.Abort() from exc
+    except CleverAgentsError as exc:
+        console.print(f"[red]Error:[/red] {exc.message}")
+        raise typer.Abort() from exc
+    except Exception as exc:
+        if isinstance(exc, (typer.Abort, typer.Exit)):
+            raise
+        console.print(f"[red]Unexpected error:[/red] {exc}")
+        raise typer.Abort() from exc
+
+
+@app.command("remove")
+def equivalence_remove(
+    virtual_id: Annotated[
+        str,
+        typer.Argument(help="Virtual resource ULID"),
+    ],
+    physical_id: Annotated[
+        str,
+        typer.Argument(help="Physical resource ULID"),
+    ],
+    yes: Annotated[
+        bool,
+        typer.Option("--yes", "-y", help="Skip confirmation prompt"),
+    ] = False,
+    fmt: Annotated[
+        str,
+        typer.Option("--format", "-f", help=_FORMAT_HELP),
+    ] = "rich",
+) -> None:
+    """Remove a virtual resource equivalence link.
+
+    Examples:
+        agents resource equivalence remove VIRT_ULID PHYS_ULID
+        agents resource equivalence remove --yes VIRT_ULID PHYS_ULID
+    """
+    _validate_ulid(virtual_id, "virtual_id")
+    _validate_ulid(physical_id, "physical_id")
+    try:
+        if not yes:
+            confirm = typer.confirm(
+                f"Remove equivalence link '{virtual_id}' -> '{physical_id}'?",
+            )
+            if not confirm:
+                console.print("[yellow]Aborted.[/yellow]")
+                raise typer.Abort()
+
+        service = _get_equivalence_service()
+        service.remove_link(virtual_id, physical_id)
+
+        if fmt != OutputFormat.RICH.value:
+            data = {
+                "status": "removed",
+                "virtual_resource_id": virtual_id,
+                "physical_resource_id": physical_id,
+            }
+            console.print(format_output(data, fmt))
+            return
+
+        console.print(
+            f"[green]Removed equivalence link:[/green] {virtual_id} -> {physical_id}"
+        )
+
+    except NotFoundError as exc:
+        console.print(f"[red]Equivalence link not found:[/red] {exc.resource_id}")
+        raise typer.Abort() from exc
+    except CleverAgentsError as exc:
+        console.print(f"[red]Error:[/red] {exc.message}")
+        raise typer.Abort() from exc
+    except Exception as exc:
+        if isinstance(exc, (typer.Abort, typer.Exit)):
+            raise
+        console.print(f"[red]Unexpected error:[/red] {exc}")
+        raise typer.Abort() from exc

src/cleveragents/domain/models/core/__init__.py

@@ -314,6 +314,12 @@ from cleveragents.domain.models.core.uko import (
     UKOVersion,
 )
 
+# Virtual resource equivalence link models
+from cleveragents.domain.models.core.virtual_resource_link import (
+    LinkType,
+    VirtualResourceLink,
+)
+
 __all__ = [
     "ASYNC_TERMINAL_STATUSES",
     "BUILTIN_PROFILES",
@@ -424,6 +430,7 @@ __all__ = [
     "InvocationTracker",
     "LegacyChangeSet",
     "LifecyclePlan",
+    "LinkType",
     "LinkedResource",
     "MaxContextCount",
     "MessageRole",
@@ -531,6 +538,7 @@ __all__ = [
     "User",
     "Validation",
     "ValidationMode",
+    "VirtualResourceLink",
     "build_provenance_map",
     "can_transition",
     "can_transition_job",

src/cleveragents/domain/models/core/virtual_resource_link.py

@@ -0,0 +1,136 @@
+"""Virtual resource equivalence link domain models.
+
+A **VirtualResourceLink** maps a virtual resource to a physical resource,
+recording the equivalence relationship between them.  Equivalence is
+determined by a computed key (content hash, name, or identity) and a
+link type that describes how the equivalence was established.
+
+## Link Types
+
+| Value          | Description                                      |
+|----------------|--------------------------------------------------|
+| `content_hash` | Equivalence by content hash (SHA-256, Merkle)    |
+| `name`         | Equivalence by name (branch name, tag name, URL) |
+| `identity`     | Identity equivalence (commit hash, tree hash)    |
+
+## Conflict Policy
+
+Physical resources take precedence over virtual resources.  When a
+physical resource diverges from its virtual parent, the link is removed
+and a divergence event is logged.  Virtual resources are never mutated
+directly — they serve as read-only identity anchors.
+
+Based on:
+
+- Specification: Physical vs Virtual Resources (lines 24383-24435)
+- Specification: Lazy Virtual Node Materialization (lines 24437-24468)
+- Specification: Equivalence linking (lines 24397-24417)
+- Issue #334: Virtual resource equivalence tracking
+"""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime
+from enum import StrEnum
+
+from pydantic import BaseModel, ConfigDict, Field, field_validator
+
+# ULID: 26 characters, Crockford's base32
+ULID_PATTERN = r"^[0-9A-HJKMNP-TV-Z]{26}$"
+
+
+class LinkType(StrEnum):
+    """How equivalence was established between virtual and physical.
+
+    | Value          | Description                                      |
+    |----------------|--------------------------------------------------|
+    | `content_hash` | Equivalence by content hash (SHA-256, Merkle)    |
+    | `name`         | Equivalence by name (branch name, tag name, URL) |
+    | `identity`     | Identity equivalence (commit hash, tree hash)    |
+    """
+
+    CONTENT_HASH = "content_hash"
+    NAME = "name"
+    IDENTITY = "identity"
+
+
+class VirtualResourceLink(BaseModel):
+    """Domain model for a virtual-to-physical resource equivalence link.
+
+    Each link records that a physical resource is a manifestation of a
+    virtual resource's abstract identity.  The ``equivalence_key`` is the
+    computed value that caused the link (e.g., a SHA-256 hash, a branch
+    name, a commit hash).
+
+    **Constraints**:
+
+    - ``id`` must match ULID format
+    - ``virtual_resource_id`` must match ULID format
+    - ``physical_resource_id`` must match ULID format
+    - ``equivalence_key`` must not be empty or whitespace-only
+    - ``confidence`` must be between 0.0 and 1.0 inclusive
+
+    Based on specification.md section "Virtual Resource Equivalence".
+    """
+
+    id: str = Field(
+        ...,
+        description="Unique ULID identifier for this link",
+        pattern=ULID_PATTERN,
+    )
+    virtual_resource_id: str = Field(
+        ...,
+        description="ULID of the virtual resource (abstract identity)",
+        pattern=ULID_PATTERN,
+    )
+    physical_resource_id: str = Field(
+        ...,
+        description="ULID of the physical resource (concrete manifestation)",
+        pattern=ULID_PATTERN,
+    )
+    equivalence_key: str = Field(
+        ...,
+        min_length=1,
+        max_length=1024,
+        description="Computed equivalence key (hash, name, or identity value)",
+    )
+    link_type: LinkType = Field(
+        default=LinkType.CONTENT_HASH,
+        description="How equivalence was established",
+    )
+    confidence: float = Field(
+        default=1.0,
+        ge=0.0,
+        le=1.0,
+        description="Confidence score for this equivalence (0.0-1.0)",
+    )
+    created_at: datetime = Field(
+        default_factory=lambda: datetime.now(tz=UTC),
+        description="When this link was created (UTC)",
+    )
+    updated_at: datetime = Field(
+        default_factory=lambda: datetime.now(tz=UTC),
+        description="When this link was last updated (UTC)",
+    )
+
+    model_config = ConfigDict(
+        frozen=True,
+        str_strip_whitespace=True,
+        use_enum_values=True,
+    )
+
+    # -- Validators ----------------------------------------------------------
+
+    @field_validator("equivalence_key")
+    @classmethod
+    def validate_equivalence_key(cls: type[VirtualResourceLink], v: str) -> str:
+        """Validate equivalence key is non-empty."""
+        if not v.strip():
+            raise ValueError("equivalence_key cannot be empty or whitespace-only")
+        return v
+
+
+__all__ = [
+    "LinkType",
+    "VirtualResourceLink",
+]

src/cleveragents/infrastructure/database/models.py

@@ -25,6 +25,7 @@ Alembic migrations.
 | ``repo_indexes``            | ``RepoIndexModel``           | Repo index metadata|
 | ``indexed_files``           | ``IndexedFileModel``         | Per-file records   |
 | ``correction_attempts``     | ``CorrectionAttemptModel``   | Correction records |
+| ``virtual_resource_links``  | ``VirtualResourceLinkModel`` | Equivalence links  |
 
 Based on ADR-007 (Repository Pattern) and Phase 0 discovery.
 Includes spec-aligned lifecycle models per Stage A5
@@ -3279,6 +3280,76 @@ class CorrectionAttemptModel(Base):  # type: ignore[misc]
         )
 
 
+# ---------------------------------------------------------------------------
+# Virtual Resource Equivalence Link Models (M7 - issue #334)
+# ---------------------------------------------------------------------------
+
+
+class VirtualResourceLinkModel(Base):  # type: ignore[misc]
+    """Database model for virtual-to-physical resource equivalence links.
+
+    Maps virtual resource ULIDs to physical resource ULIDs with
+    uniqueness constraints.  Used by ``ResourceEquivalenceService``
+    to track which physical resources are manifestations of the same
+    virtual identity.
+
+    Table: ``virtual_resource_links``
+    """
+
+    __allow_unmapped__ = True
+    __tablename__ = "virtual_resource_links"
+
+    # PK: ULID (26-char string)
+    id = Column(String(26), primary_key=True)
+
+    # FK to resources (virtual)
+    virtual_resource_id = Column(
+        String(26),
+        ForeignKey("resources.resource_id", ondelete="CASCADE"),
+        nullable=False,
+    )
+
+    # FK to resources (physical)
+    physical_resource_id = Column(
+        String(26),
+        ForeignKey("resources.resource_id", ondelete="CASCADE"),
+        nullable=False,
+    )
+
+    # Computed equivalence key (hash, name, or identity value)
+    equivalence_key = Column(String(1024), nullable=False)
+
+    # How equivalence was established
+    link_type = Column(String(30), nullable=False, default="content_hash")
+
+    # Confidence score (0.0 - 1.0)
+    confidence = Column(Float, nullable=False, default=1.0)
+
+    # Timestamps (ISO-8601 strings, up to 40 chars with tz+microseconds)
+    created_at = Column(String(40), nullable=False)
+    updated_at = Column(String(40), nullable=False)
+
+    __table_args__ = (
+        UniqueConstraint(
+            "virtual_resource_id",
+            "physical_resource_id",
+            name="uq_virtual_resource_links_virtual_physical",
+        ),
+        CheckConstraint(
+            "link_type IN ('content_hash', 'name', 'identity')",
+            name="ck_virtual_resource_links_link_type",
+        ),
+        CheckConstraint(
+            "confidence >= 0.0 AND confidence <= 1.0",
+            name="ck_virtual_resource_links_confidence",
+        ),
+        Index("ix_virtual_resource_links_virtual", "virtual_resource_id"),
+        Index("ix_virtual_resource_links_physical", "physical_resource_id"),
+        Index("ix_virtual_resource_links_equiv_key", "equivalence_key"),
+        Index("ix_virtual_resource_links_link_type", "link_type"),
+    )
+
+
 # Database initialization functions
 def init_database(database_url: str = "sqlite:///.cleveragents/db.sqlite") -> Any:
     """Initialize the database.