diff --git a/docs/concepts/index.md b/docs/concepts/index.md
index 48504f68..5a066352 100644
--- a/docs/concepts/index.md
+++ b/docs/concepts/index.md
@@ -417,6 +417,8 @@ classDiagram
     class IngestionParams {
         +clear_data: bool
         +n_cores: int
+        +resources: list[str]?
+        +vertices: list[str]?
         +batch_size: int
         +max_items: int?
         +dry: bool
diff --git a/docs/examples/example-5.md b/docs/examples/example-5.md
index 8dc0b01a..b2c100be 100644
--- a/docs/examples/example-5.md
+++ b/docs/examples/example-5.md
@@ -776,6 +776,38 @@ bindings = Bindings(
 )
 ```
 
+### Runtime proxy wiring (no secrets in YAML)
+
+Once your `bindings` block contains `connector_connection` proxy labels, you must
+register the real runtime connection config under each `conn_proxy` and bind the
+manifest connectors to that proxy:
+
+```python
+from graflo.hq.connection_provider import (
+    InMemoryConnectionProvider,
+    PostgresGeneralizedConnConfig,
+)
+from graflo.hq import IngestionParams
+
+provider = InMemoryConnectionProvider()
+provider.register_generalized_config(
+    conn_proxy="postgres_source",
+    config=PostgresGeneralizedConnConfig(config=postgres_conf),
+)
+provider.bind_from_bindings(bindings=bindings)
+
+engine.define_and_ingest(
+    manifest=manifest.model_copy(update={"bindings": bindings}),
+    target_db_config=conn_conf,
+    ingestion_params=IngestionParams(clear_data=True),
+    recreate_schema=False,
+    connection_provider=provider,
+)
+```
+
+For the common single-proxy case you can replace the `register_generalized_config(...)` +
+`bind_from_bindings(...)` steps with `provider.bind_single_config_for_bindings(...)`.
+
 ## Viewing Results in Graph Database Web Interfaces
 
 After successful ingestion, you can explore your graph data using each database's web interface. The default ports and access information are listed below. Check the corresponding `.env` files in the `docker/` directories for custom port configurations.
diff --git a/docs/examples/example-8.md b/docs/examples/example-8.md
new file mode 100644
index 00000000..f562111b
--- /dev/null
+++ b/docs/examples/example-8.md
@@ -0,0 +1,25 @@
+# Example 8: Multi-Edge Weights with Filters and `dress` Transforms
+
+This example ingests ticker CSV data into Neo4j with:
+
+- **Two vertex types** — `ticker` (by `oftic`) and `metric` (by `name` + `value`), where `metric` rows are filtered so only Open, Close, and Volume with positive values become vertices.
+- **One edge** — `ticker` → `metric` with **multiple weights** (`direct` on `t_obs` plus nested `vertices` metadata on the metric endpoint).
+- **Transforms with `dress`** — `round_str` and `int` transforms targeted at specific `(name, value)` pairs via `dress: { key: name, value: value }`, plus a date parse that emits `t_obs`.
+
+## Layout
+
+- `examples/8-multi-edges-weights/manifest.yaml` — logical schema, DB profile (Neo4j indexes, edge specs), transforms, and `ticker_data` resource pipeline.
+- `examples/8-multi-edges-weights/ingest.py` — `FileConnector` + `Bindings`, then `GraphEngine.define_and_ingest(...)`.
+- `examples/8-multi-edges-weights/data.csv` — sample OHLCV-style rows.
+
+## Run locally
+
+From the example directory, with Neo4j running (see repo `docker/neo4j`), run:
+
+```bash
+uv run python ingest.py
+```
+
+## Related
+
+- [Polymorphic routing (Example 7)](example-7.md) uses `vertex_router` / `edge_router` for type-discriminated tables; this example uses **filters** on a vertex type and **multi-weight** edges instead.
diff --git a/docs/examples/example-9.md b/docs/examples/example-9.md
new file mode 100644
index 00000000..8c6f5997
--- /dev/null
+++ b/docs/examples/example-9.md
@@ -0,0 +1,64 @@
+# Example 9: Explicit `connector_connection` Proxy Wiring
+
+This example shows the full proxy chain end-to-end:
+
+`Resource -> Connector -> ConnectionProxy -> RuntimeConnectionConfig`
+
+The manifest stays credential-free: `bindings.connector_connection` only contains proxy labels (`conn_proxy`). The script then registers the real connection config at runtime.
+
+## Manifest: what `connector_connection` looks like
+
+Inside `bindings` you explicitly map each connector to a proxy label:
+
+```yaml
+bindings:
+  connector_connection:
+    - connector: users
+      conn_proxy: postgres_source
+    - connector: products
+      conn_proxy: postgres_source
+    - connector: purchases
+      conn_proxy: postgres_source
+    - connector: follows
+      conn_proxy: postgres_source
+```
+
+In the code, connectors omit `connector.name` and use `connector.resource_name` (so the manifest references are stable and human-readable).
+
+## Runtime: how the proxy label becomes a real DB config
+
+The script wires runtime config and binds the manifest connectors to the chosen proxy:
+
+```python
+from graflo.hq.connection_provider import (
+    InMemoryConnectionProvider,
+    PostgresGeneralizedConnConfig,
+)
+
+provider = InMemoryConnectionProvider()
+
+provider.register_generalized_config(
+    conn_proxy="postgres_source",
+    config=PostgresGeneralizedConnConfig(config=postgres_conf),
+)
+
+provider.bind_from_bindings(bindings=bindings)
+```
+
+For the common single-DB / single-proxy case, you can also use:
+
+```python
+provider.bind_single_config_for_bindings(
+    bindings=bindings,
+    conn_proxy="postgres_source",
+    config=PostgresGeneralizedConnConfig(config=postgres_conf),
+)
+```
+
+## Full script
+
+See:
+
+- `examples/9-connector-connection-proxy/explicit_proxy_binding.py`
+- `examples/9-connector-connection-proxy/README.md`
+
diff --git a/docs/examples/index.md b/docs/examples/index.md
index 826e2d36..47fc754e 100644
--- a/docs/examples/index.md
+++ b/docs/examples/index.md
@@ -6,4 +6,6 @@
 4. [Neo4j Ingestion with Dynamic Relations from Keys](example-4.md)
 5. **[🚀 PostgreSQL Schema Inference and Ingestion](example-5.md)** - **Automatically infer graph schemas from normalized PostgreSQL databases (3NF)** with proper primary keys (PK) and foreign keys (FK). Uses intelligent heuristics to detect vertices and edges - no manual schema definition needed! Perfect for migrating relational data to graph databases.
 6. **[🔗 RDF / Turtle Ingestion with Explicit Resource Mapping](example-6.md)** - **Infer graph schemas from OWL ontologies and ingest RDF data** using explicit `SparqlConnector` resource mapping. Supports local Turtle files and remote SPARQL endpoints. Perfect for knowledge graph pipelines built on semantic web standards.
-7. **[Polymorphic Objects and Relations](example-7.md)** — **Route polymorphic entities and dynamic relations** using `vertex_router` and `edge_router`. One objects table (Person, Vehicle, Institution) and one relations table (EMPLOYED_BY, OWNS, FUNDS, etc.) map to a rich graph with type discriminators and `relation_map`.
\ No newline at end of file
+7. **[Polymorphic Objects and Relations](example-7.md)** — **Route polymorphic entities and dynamic relations** using `vertex_router` and `edge_router`. One objects table (Person, Vehicle, Institution) and one relations table (EMPLOYED_BY, OWNS, FUNDS, etc.) map to a rich graph with type discriminators and `relation_map`.
+8. **[Multi-Edge Weights with Filters and `dress` Transforms](example-8.md)** — **Ticker-style CSV to Neo4j** with vertex filters, multiple edge weights, and `dress`-scoped transforms on metric name/value pairs.
+9. **[Explicit `connector_connection` Proxy Wiring](example-9.md)** — Show how manifest proxy labels (`conn_proxy`) are resolved at runtime into real DB configs via `ConnectionProvider`.
\ No newline at end of file
diff --git a/docs/getting_started/creating_manifest.md b/docs/getting_started/creating_manifest.md
index 72f17d36..04512145 100644
--- a/docs/getting_started/creating_manifest.md
+++ b/docs/getting_started/creating_manifest.md
@@ -91,6 +91,30 @@ The block can be left empty in-file (`bindings: {}`) and supplied at runtime for
 
 Use `bindings` for **where data comes from** (and optionally **which proxy label** supplies runtime credentials for each SQL/SPARQL connector).
 
+### Runtime proxy wiring (example)
+
+The manifest contains proxy labels only. At runtime you register the real connection config and bind manifest connectors to those proxy labels:
+
+```python
+from graflo.hq.connection_provider import (
+    InMemoryConnectionProvider,
+    PostgresGeneralizedConnConfig,
+)
+
+provider = InMemoryConnectionProvider()
+provider.bind_single_config_for_bindings(
+    bindings=bindings,
+    conn_proxy="postgres_source",
+    config=PostgresGeneralizedConnConfig(config=postgres_conf),
+)
+
+engine.define_and_ingest(
+    manifest=manifest,
+    target_db_config=target_db_config,
+    connection_provider=provider,
+)
+```
+
 ## Authoring tips
 
 - Keep resource names unique across `ingestion_model.resources`.
diff --git a/docs/getting_started/quickstart.md b/docs/getting_started/quickstart.md
index a7246870..fb296f17 100644
--- a/docs/getting_started/quickstart.md
+++ b/docs/getting_started/quickstart.md
@@ -135,6 +135,30 @@ Here `schema` defines the logical graph, while `ingestion_model` defines resourc
 
 For SQL and SPARQL sources, add **`connector_connection`**: a list of `{"connector": "<connector name or hash>", "conn_proxy": "<label>"}`. At runtime, register each `conn_proxy` on an `InMemoryConnectionProvider` (or your own `ConnectionProvider`) with `GeneralizedConnConfig`. `GraphEngine` / `ResourceMapper` call `bind_connector_to_conn_proxy` when building bindings from Postgres or RDF workflows so HQ and the manifest stay aligned.
 
+### Single-DB quick path (one proxy label)
+
+When all SQL connectors use the same `conn_proxy`, you can wire the runtime config in one call:
+
+```python
+from graflo.hq.connection_provider import (
+    InMemoryConnectionProvider,
+    PostgresGeneralizedConnConfig,
+)
+
+provider = InMemoryConnectionProvider()
+provider.bind_single_config_for_bindings(
+    bindings=bindings,
+    conn_proxy="postgres_source",
+    config=PostgresGeneralizedConnConfig(config=postgres_conf),
+)
+
+engine.define_and_ingest(
+    manifest=manifest,
+    target_db_config=conn_conf,
+    connection_provider=provider,
+)
+```
+
 The `ingest()` method takes:
 - `target_db_config`: Target graph database configuration (where to write the graph)
 - `bindings`: Source data connectors (where to read data from - files or database tables)
diff --git a/docs/reference/data_source/index.md b/docs/reference/data_source/index.md
index d900d05b..e43b10d1 100644
--- a/docs/reference/data_source/index.md
+++ b/docs/reference/data_source/index.md
@@ -172,6 +172,16 @@ caster = Caster(schema=schema, ingestion_model=ingestion_model)
 ingestion_params = IngestionParams(
     batch_size=1000,  # Process 1000 items per batch
     clear_data=False,
+    # Optional: restrict ingestion scope.
+    # `resources` limits which logical resources (from the manifest) are processed.
+    # `vertices` is an allow-list of vertex types to ingest.
+    #
+    # Implementation detail: the allow-list is applied early in the runtime actor
+    # pipeline (not just as a late DB-write filter). Disallowed vertex types are
+    # not extracted/assembled, and edges are only emitted when both endpoints
+    # are allowed.
+    resources=["your_resource_name"],
+    vertices=["your_vertex_name"],
 )
 
 asyncio.run(
diff --git a/docs/reference/db/cypher/__init__.md b/docs/reference/db/cypher/__init__.md
new file mode 100644
index 00000000..3c72f280
--- /dev/null
+++ b/docs/reference/db/cypher/__init__.md
@@ -0,0 +1,3 @@
+# `graflo.db.cypher`
+
+::: graflo.db.cypher
diff --git a/docs/reference/db/cypher/escape.md b/docs/reference/db/cypher/escape.md
new file mode 100644
index 00000000..84c3e8d7
--- /dev/null
+++ b/docs/reference/db/cypher/escape.md
@@ -0,0 +1,3 @@
+# `graflo.db.cypher.escape`
+
+::: graflo.db.cypher.escape
diff --git a/docs/reference/db/cypher/rel_merge.md b/docs/reference/db/cypher/rel_merge.md
new file mode 100644
index 00000000..fcbe6423
--- /dev/null
+++ b/docs/reference/db/cypher/rel_merge.md
@@ -0,0 +1,3 @@
+# `graflo.db.cypher.rel_merge`
+
+::: graflo.db.cypher.rel_merge
diff --git a/docs/reference/hq/ingestion_parameters.md b/docs/reference/hq/ingestion_parameters.md
new file mode 100644
index 00000000..c9c8e035
--- /dev/null
+++ b/docs/reference/hq/ingestion_parameters.md
@@ -0,0 +1,3 @@
+# `graflo.hq.ingestion_parameters`
+
+::: graflo.hq.ingestion_parameters
diff --git a/docs/reference/index.md b/docs/reference/index.md
index 006668e0..a99a405c 100644
--- a/docs/reference/index.md
+++ b/docs/reference/index.md
@@ -28,6 +28,7 @@ Database connection and management components:
   - [Utilities](db/arango/util.md): ArangoDB-specific utility functions
 - **[Neo4j](db/neo4j/__init__.md)**:
   - [Connection](db/neo4j/conn.md): Neo4j-specific connection implementation
+- **[Cypher helpers](db/cypher/__init__.md)** — Shared Cypher utilities ([escape](db/cypher/escape.md), [relationship MERGE](db/cypher/rel_merge.md))
 - **[FalkorDB](db/falkordb/__init__.md)**:
   - [Connection](db/falkordb/conn.md): FalkorDB-specific connection implementation
 - **[TigerGraph](db/tigergraph/__init__.md)**:
@@ -48,6 +49,7 @@ Database connection and management components:
 Main graflo functionality:
 
 - **[Caster](hq/caster.md)**: Main data ingestion and transformation engine
+- **[Ingestion parameters](hq/ingestion_parameters.md)**: `IngestionParams`, row-error policy types, and batch cast results (`CastBatchResult`, …)
 - **[Data Sources](data_source/index.md)**: Data source abstraction layer (files, APIs, SQL, in-memory)
 - **[Ontology](onto.md)**: Core data types and enums
 
diff --git a/examples/5-ingest-postgres/generated-manifest.yaml b/examples/5-ingest-postgres/generated-manifest.yaml
index 7b36be97..cf205276 100644
--- a/examples/5-ingest-postgres/generated-manifest.yaml
+++ b/examples/5-ingest-postgres/generated-manifest.yaml
@@ -1,9 +1,4 @@
-db_profile:
-    db_flavor: tigergraph
-    vertex_storage_names:
-        products: products
-        users: users
-graph:
+core_schema:
     edge_config:
         edges:
         -   relation: follows
@@ -52,5 +47,10 @@ graph:
             identity:
             -   id
             name: users
+db_profile:
+    db_flavor: tigergraph
+    vertex_storage_names:
+        products: products
+        users: users
 metadata:
     name: accounting
diff --git a/examples/9-connector-connection-proxy/README.md b/examples/9-connector-connection-proxy/README.md
new file mode 100644
index 00000000..cdd1dc07
--- /dev/null
+++ b/examples/9-connector-connection-proxy/README.md
@@ -0,0 +1,13 @@
+# Example 9: Explicit connector_connection proxy wiring
+
+This example demonstrates the non-secret runtime indirection:
+
+`Resource -> Connector -> ConnectionProxy -> RuntimeConnectionConfig`
+
+Key points:
+- The manifest stores only `conn_proxy` labels inside `bindings.connector_connection`.
+- The runtime script registers the real `PostgresConfig` under that proxy label
+  via `InMemoryConnectionProvider`.
+- `provider.bind_from_bindings(bindings=...)` connects manifest connectors
+  to the proxy label so ingestion can resolve `conn_proxy -> config`.
+
diff --git a/examples/9-connector-connection-proxy/explicit_proxy_binding.py b/examples/9-connector-connection-proxy/explicit_proxy_binding.py
new file mode 100644
index 00000000..a00d6f6a
--- /dev/null
+++ b/examples/9-connector-connection-proxy/explicit_proxy_binding.py
@@ -0,0 +1,149 @@
+"""
+Example 9: Explicit connector_connection proxy wiring
+
+This example focuses on the runtime indirection:
+
+Resource -> Connector -> ConnectionProxy -> RuntimeConnectionConfig
+
+Unlike `examples/5-ingest-postgres`, we explicitly author:
+- `Bindings.connector_connection` in the manifest (proxy labels, no secrets)
+- the runtime `InMemoryConnectionProvider` wiring that registers proxy configs
+  and binds manifest connectors to those proxies.
+"""
+
+from __future__ import annotations
+
+import logging
+from pathlib import Path
+
+from graflo.architecture.contract.bindings import Bindings, TableConnector
+from graflo.hq import GraphEngine, IngestionParams
+from graflo.hq.connection_provider import (
+    InMemoryConnectionProvider,
+    PostgresGeneralizedConnConfig,
+)
+from graflo.db.postgres.util import load_schema_from_sql_file
+from graflo.db import PostgresConfig, TigergraphConfig
+
+logger = logging.getLogger(__name__)
+
+
+def _load_mock_postgres_schema(*, postgres_conf: PostgresConfig) -> None:
+    schema_file = (
+        Path(__file__).resolve().parents[1]
+        / "5-ingest-postgres"
+        / "data"
+        / "mock_schema.sql"
+    )
+    if not schema_file.exists():
+        logger.warning("Mock schema file not found: %s", schema_file)
+        return
+
+    load_schema_from_sql_file(
+        config=postgres_conf,
+        schema_file=schema_file,
+        continue_on_error=True,  # e.g. DROP TABLE IF EXISTS ...
+    )
+
+
+def make_explicit_postgres_bindings(conn_proxy: str) -> Bindings:
+    """Create manifest bindings with explicit connector_connection proxy labels."""
+    # In this example we keep `connector.name` omitted and rely on
+    # connector.resource_name as the stable manifest alias.
+    connectors = [
+        TableConnector(
+            table_name="users",
+            schema_name="public",
+            resource_name="users",
+        ),
+        TableConnector(
+            table_name="products",
+            schema_name="public",
+            resource_name="products",
+        ),
+        TableConnector(
+            table_name="purchases",
+            schema_name="public",
+            resource_name="purchases",
+        ),
+        TableConnector(
+            table_name="follows",
+            schema_name="public",
+            resource_name="follows",
+        ),
+    ]
+
+    connector_connection = [
+        {"connector": "users", "conn_proxy": conn_proxy},
+        {"connector": "products", "conn_proxy": conn_proxy},
+        {"connector": "purchases", "conn_proxy": conn_proxy},
+        {"connector": "follows", "conn_proxy": conn_proxy},
+    ]
+
+    return Bindings(
+        connectors=connectors,
+        # We don't need `resource_connector` because each connector defines
+        # `resource_name`, which Bindings uses to map resources -> connectors.
+        resource_connector=[],
+        connector_connection=connector_connection,
+    )
+
+
+def main() -> None:
+    logging.basicConfig(level=logging.WARNING, handlers=[logging.StreamHandler()])
+    logging.getLogger("graflo").setLevel(logging.INFO)
+
+    # Step 1: Connect to PostgreSQL (source database)
+    postgres_conf = PostgresConfig.from_docker_env()
+
+    # Optional: initialize a mock schema (for quick local experimentation)
+    _load_mock_postgres_schema(postgres_conf=postgres_conf)
+
+    # Step 2: Connect to target graph database
+    conn_conf = TigergraphConfig.from_docker_env()
+    db_type = conn_conf.connection_type
+
+    # Step 3: Infer schema (optional for this example; bindings are explicit below)
+    engine = GraphEngine(target_db_flavor=db_type)
+    manifest = engine.infer_manifest(
+        postgres_conf, schema_name="public", fuzzy_threshold=0.8
+    )
+    schema = manifest.require_schema()
+    schema.metadata.name = "accounting"
+    ingestion_model = manifest.require_ingestion_model()
+    _ = ingestion_model  # kept to emphasize the manifest contains ingestion resources
+
+    # Step 4: Author bindings with explicit connector_connection proxy labels
+    conn_proxy = "postgres_source"
+    bindings = make_explicit_postgres_bindings(conn_proxy=conn_proxy)
+
+    manifest = manifest.model_copy(update={"bindings": bindings})
+    manifest.finish_init()
+
+    # Step 5: Wire runtime proxy config + bind manifest connectors to the proxy
+    # Manifest contains proxy labels only; register the real connection config here.
+    provider = InMemoryConnectionProvider()
+    provider.register_generalized_config(
+        conn_proxy=conn_proxy,
+        config=PostgresGeneralizedConnConfig(config=postgres_conf),
+    )
+    provider.bind_from_bindings(bindings=bindings)
+
+    # Step 6: Define schema and ingest (runtime uses `connection_provider=provider`)
+    engine.define_and_ingest(
+        manifest=manifest,
+        target_db_config=conn_conf,
+        ingestion_params=IngestionParams(clear_data=True),
+        recreate_schema=True,
+        connection_provider=provider,
+    )
+
+    print("\n" + "=" * 80)
+    print("Explicit proxy binding ingestion complete!")
+    print("=" * 80)
+    print(f"Schema: {schema.metadata.name}")
+    print(f"Bindings.connector_connection: {manifest.bindings.connector_connection}")
+
+
+if __name__ == "__main__":
+    main()
diff --git a/graflo/architecture/contract/declarations/ingestion_model/model.py b/graflo/architecture/contract/declarations/ingestion_model/model.py
index ecdfded8..ddc39073 100644
--- a/graflo/architecture/contract/declarations/ingestion_model/model.py
+++ b/graflo/architecture/contract/declarations/ingestion_model/model.py
@@ -73,6 +73,7 @@ def finish_init(
         *,
         strict_references: bool = False,
         dynamic_edge_feedback: bool = False,
+        allowed_vertex_names: set[str] | None = None,
     ) -> None:
         """Initialize resources against graph model and transform library."""
         self._rebuild_runtime_state()
@@ -83,6 +84,7 @@ def finish_init(
                 transforms=self._transforms,
                 strict_references=strict_references,
                 dynamic_edge_feedback=dynamic_edge_feedback,
+                allowed_vertex_names=allowed_vertex_names,
             )
 
     def _rebuild_runtime_state(self) -> None:
diff --git a/graflo/architecture/contract/declarations/resource.py b/graflo/architecture/contract/declarations/resource.py
index c0374a2c..4004abbd 100644
--- a/graflo/architecture/contract/declarations/resource.py
+++ b/graflo/architecture/contract/declarations/resource.py
@@ -84,6 +84,34 @@ def _resolve_type_caster(name: str) -> Callable[..., Any] | None:
     return None
 
 
+def _filter_vertex_config_by_allowed(
+    vertex_config: VertexConfig,
+    *,
+    allowed_vertex_names: set[str] | None,
+) -> VertexConfig:
+    """Derive a filtered VertexConfig for runtime actor execution.
+
+    This intentionally filters only the vertex collections present in
+    *allowed_vertex_names*; it does not attempt to rewrite edge configs.
+    """
+    if allowed_vertex_names is None:
+        return vertex_config
+
+    allowed = allowed_vertex_names
+    filtered_vertices = [v for v in vertex_config.vertices if v.name in allowed]
+    filtered_blank_vertices = [b for b in vertex_config.blank_vertices if b in allowed]
+    filtered_force_types = {
+        name: types
+        for name, types in vertex_config.force_types.items()
+        if name in allowed
+    }
+    return VertexConfig(
+        vertices=filtered_vertices,
+        blank_vertices=filtered_blank_vertices,
+        force_types=filtered_force_types,
+    )
+
+
 class EdgeInferSpec(ConfigBaseModel):
     """Selector for controlling inferred edge emission."""
 
@@ -251,6 +279,7 @@ def finish_init(
         *,
         strict_references: bool = False,
         dynamic_edge_feedback: bool = False,
+        allowed_vertex_names: set[str] | None = None,
     ) -> None:
         """Complete resource initialization.
 
@@ -268,6 +297,7 @@ def finish_init(
             transforms=transforms,
             strict_references=strict_references,
             dynamic_edge_feedback=dynamic_edge_feedback,
+            allowed_vertex_names=allowed_vertex_names,
         )
 
     def _edge_ids_from_edge_actors(self) -> set[EdgeId]:
@@ -318,9 +348,15 @@ def _rebuild_runtime(
         transforms: dict[str, ProtoTransform],
         strict_references: bool = False,
         dynamic_edge_feedback: bool = False,
+        allowed_vertex_names: set[str] | None = None,
     ) -> None:
         """Rebuild runtime actor initialization state from typed context."""
-        object.__setattr__(self, "_vertex_config", vertex_config)
+        # Keep the full schema vertex_config for correctness validations, but
+        # use the filtered runtime vertex_config for actor execution.
+        runtime_vertex_config = _filter_vertex_config_by_allowed(
+            vertex_config, allowed_vertex_names=allowed_vertex_names
+        )
+        object.__setattr__(self, "_vertex_config", runtime_vertex_config)
         # Runtime actors may register dynamic edges; keep per-resource edge state.
         local_edge_config = EdgeConfig.model_validate(
             edge_config.to_dict(skip_defaults=False)
@@ -340,9 +376,10 @@ def _rebuild_runtime(
 
         logger.debug("total resource actor count : %s", self.root.count())
         init_ctx = ActorInitContext(
-            vertex_config=vertex_config,
+            vertex_config=runtime_vertex_config,
             edge_config=self._edge_config,
             transforms=transforms,
+            allowed_vertex_names=allowed_vertex_names,
             infer_edges=self.infer_edges,
             infer_edge_only={spec.edge_id for spec in self.infer_edge_only},
             infer_edge_except=infer_edge_except,
diff --git a/graflo/architecture/pipeline/runtime/actor/base.py b/graflo/architecture/pipeline/runtime/actor/base.py
index e65e5ef6..5de03d09 100644
--- a/graflo/architecture/pipeline/runtime/actor/base.py
+++ b/graflo/architecture/pipeline/runtime/actor/base.py
@@ -25,6 +25,7 @@ class ActorInitContext:
     vertex_config: VertexConfig
     edge_config: EdgeConfig
     transforms: dict[str, ProtoTransform]
+    allowed_vertex_names: set[str] | None = None
     infer_edges: bool = True
     infer_edge_only: set[EdgeId] = field(default_factory=set)
     infer_edge_except: set[EdgeId] = field(default_factory=set)
diff --git a/graflo/architecture/pipeline/runtime/actor/edge.py b/graflo/architecture/pipeline/runtime/actor/edge.py
index 691fdeca..61500c22 100644
--- a/graflo/architecture/pipeline/runtime/actor/edge.py
+++ b/graflo/architecture/pipeline/runtime/actor/edge.py
@@ -18,6 +18,7 @@ def __init__(self, config: EdgeActorConfig):
         kwargs.pop("type", None)
         self.edge = Edge.from_dict(kwargs)
         self.vertex_config: Any = None
+        self.allowed_vertex_names: set[str] | None = None
 
     @classmethod
     def from_config(cls, config: EdgeActorConfig) -> EdgeActor:
@@ -32,6 +33,7 @@ def fetch_important_items(self) -> dict[str, Any]:
 
     def finish_init(self, init_ctx: ActorInitContext) -> None:
         self.vertex_config = init_ctx.vertex_config
+        self.allowed_vertex_names = init_ctx.allowed_vertex_names
         if self.vertex_config is not None:
             init_ctx.edge_config.update_edges(
                 self.edge, vertex_config=self.vertex_config
@@ -40,6 +42,22 @@ def finish_init(self, init_ctx: ActorInitContext) -> None:
     def __call__(
         self, ctx: ExtractionContext, lindex: LocationIndex, *nargs: Any, **kwargs: Any
     ) -> ExtractionContext:
+        # Early-exit for disallowed vertex endpoints.
+        if self.allowed_vertex_names is not None and (
+            self.edge.source not in self.allowed_vertex_names
+            or self.edge.target not in self.allowed_vertex_names
+        ):
+            return ctx
+        if (
+            self.allowed_vertex_names is None
+            and self.vertex_config is not None
+            and (
+                self.edge.source not in self.vertex_config.vertex_set
+                or self.edge.target not in self.vertex_config.vertex_set
+            )
+        ):
+            return ctx
+
         ctx.edge_requests.append((self.edge, lindex))
         ctx.record_edge_intent(edge=self.edge, location=lindex)
         return ctx
diff --git a/graflo/architecture/pipeline/runtime/actor/vertex.py b/graflo/architecture/pipeline/runtime/actor/vertex.py
index 0ef816a2..017d306e 100644
--- a/graflo/architecture/pipeline/runtime/actor/vertex.py
+++ b/graflo/architecture/pipeline/runtime/actor/vertex.py
@@ -27,6 +27,7 @@ def __init__(self, config: VertexActorConfig):
             tuple(config.keep_fields) if config.keep_fields else None
         )
         self.vertex_config: VertexConfig
+        self.allowed_vertex_names: set[str] | None = None
 
     @classmethod
     def from_config(cls, config: VertexActorConfig) -> VertexActor:
@@ -37,6 +38,7 @@ def fetch_important_items(self) -> dict[str, Any]:
 
     def finish_init(self, init_ctx: ActorInitContext) -> None:
         self.vertex_config = init_ctx.vertex_config
+        self.allowed_vertex_names = init_ctx.allowed_vertex_names
 
     def _filter_and_aggregate_vertex_docs(
         self, docs: list[dict[str, Any]], doc: dict[str, Any]
@@ -123,6 +125,20 @@ def __call__(
         self, ctx: ExtractionContext, lindex: LocationIndex, *nargs: Any, **kwargs: Any
     ) -> ExtractionContext:
         doc: dict[str, Any] = kwargs.get("doc", {})
+
+        # Early-exit for disallowed vertex types.
+        # This must happen before any ctx.acc_vertex[...] access.
+        if (
+            self.allowed_vertex_names is not None
+            and self.name not in self.allowed_vertex_names
+        ):
+            return ctx
+        if (
+            self.allowed_vertex_names is None
+            and self.name not in self.vertex_config.vertex_set
+        ):
+            return ctx
+
         vertex_keys_list = self.vertex_config.fields_names(self.name)
         vertex_keys: tuple[str, ...] = tuple(vertex_keys_list)
 
diff --git a/graflo/architecture/pipeline/runtime/actor/wrapper.py b/graflo/architecture/pipeline/runtime/actor/wrapper.py
index def924f6..3f52fb8a 100644
--- a/graflo/architecture/pipeline/runtime/actor/wrapper.py
+++ b/graflo/architecture/pipeline/runtime/actor/wrapper.py
@@ -109,6 +109,7 @@ def from_config(cls, config: ActorConfig) -> ActorWrapper:
             vertex_config=VertexConfig(vertices=[]),
             edge_config=EdgeConfig(),
             transforms={},
+            allowed_vertex_names=None,
             infer_edges=True,
             infer_edge_only=set(),
             infer_edge_except=set(),
diff --git a/graflo/db/arango/conn.py b/graflo/db/arango/conn.py
index 2504db3d..9b364ba9 100644
--- a/graflo/db/arango/conn.py
+++ b/graflo/db/arango/conn.py
@@ -639,6 +639,9 @@ def _graph_name(graph_item: Any) -> str | None:
             logger.info([])
 
         if delete_all:
+            logger.warning(
+                "delete_graph_structure(delete_all=True) will remove all non-system ArangoDB graphs and collections in the selected database"
+            )
             collections_result = self.conn.collections()
             graphs_result = self.conn.graphs()
             cnames = []
diff --git a/graflo/db/conn.py b/graflo/db/conn.py
index 23878d25..247393a4 100644
--- a/graflo/db/conn.py
+++ b/graflo/db/conn.py
@@ -216,7 +216,8 @@ def delete_graph_structure(
         Args:
             vertex_types: Vertex type names to delete (database-specific interpretation)
             graph_names: Graph/database names to delete
-            delete_all: If True, delete all graphs and their associated structures
+            delete_all: If True, delete all targeted graph structures.
+                This is destructive and should only be used with explicit intent.
         """
         pass
 
diff --git a/graflo/db/falkordb/conn.py b/graflo/db/falkordb/conn.py
index 1e147305..4e413f69 100644
--- a/graflo/db/falkordb/conn.py
+++ b/graflo/db/falkordb/conn.py
@@ -446,7 +446,10 @@ def delete_graph_structure(
             graph_names: Graph names to delete entirely
             delete_all: If True, delete all nodes and relationships
         """
-        if delete_all or (not vertex_types and not graph_names):
+        if delete_all:
+            logger.warning(
+                "delete_graph_structure(delete_all=True) will remove all nodes and relationships in the selected FalkorDB graph"
+            )
             # Delete all nodes and relationships in current graph
             try:
                 self.execute("MATCH (n) DETACH DELETE n")
@@ -543,9 +546,12 @@ def init_db(self, schema: Schema, recreate_schema: bool) -> None:
     def clear_data(self, schema: Schema) -> None:
         """Remove all data from the graph without dropping the schema.
 
-        Deletes all nodes and relationships; labels (schema) remain.
+        Deletes nodes and relationships for schema-managed labels only.
         """
-        self.delete_graph_structure(delete_all=True)
+        vc = schema.resolve_db_aware(DBType.FALKORDB).vertex_config
+        vertex_types = tuple(vc.vertex_dbname(v) for v in vc.vertex_set)
+        if vertex_types:
+            self.delete_graph_structure(vertex_types=vertex_types)
 
     def upsert_docs_batch(
         self,
diff --git a/graflo/db/memgraph/conn.py b/graflo/db/memgraph/conn.py
index 923deb78..145e258f 100644
--- a/graflo/db/memgraph/conn.py
+++ b/graflo/db/memgraph/conn.py
@@ -628,9 +628,12 @@ def init_db(self, schema: Schema, recreate_schema: bool) -> None:
     def clear_data(self, schema: Schema) -> None:
         """Remove all data from the graph without dropping the schema.
 
-        Deletes all nodes and relationships; labels (schema) remain.
+        Deletes nodes and relationships for schema-managed labels only.
         """
-        self.delete_graph_structure(delete_all=True)
+        vc = schema.resolve_db_aware(DBType.MEMGRAPH).vertex_config
+        vertex_types = tuple(vc.vertex_dbname(v) for v in vc.vertex_set)
+        if vertex_types:
+            self.delete_graph_structure(vertex_types=vertex_types)
 
     def upsert_docs_batch(
         self,
diff --git a/graflo/db/nebula/conn.py b/graflo/db/nebula/conn.py
index 091c48c9..51517da3 100644
--- a/graflo/db/nebula/conn.py
+++ b/graflo/db/nebula/conn.py
@@ -447,7 +447,9 @@ def init_db(self, schema: Schema, recreate_schema: bool) -> None:
     # ------------------------------------------------------------------
 
     def clear_data(self, schema: Schema) -> None:
-        for vname in schema.core_schema.vertex_config.vertex_set:
+        vc = schema.resolve_db_aware(DBType.NEBULA).vertex_config
+        for vertex in vc.vertex_set:
+            vname = vc.vertex_dbname(vertex)
             try:
                 self._execute(
                     f"LOOKUP ON `{vname}` YIELD id(vertex) AS vid "
diff --git a/graflo/db/neo4j/conn.py b/graflo/db/neo4j/conn.py
index 4eaaf0dc..e9cd2ae8 100644
--- a/graflo/db/neo4j/conn.py
+++ b/graflo/db/neo4j/conn.py
@@ -359,11 +359,19 @@ def delete_graph_structure(
         cnames = vertex_types
         if cnames:
             for c in cnames:
-                q = f"MATCH (n:{c}) DELETE n"
+                q = f"MATCH (n:{c}) DETACH DELETE n"
                 self.execute(q)
-        else:
+            return
+
+        if delete_all:
+            logger.warning(
+                "delete_graph_structure(delete_all=True) will remove all nodes and relationships in the selected Neo4j database"
+            )
             self._drop_all_user_indexes_and_constraints()
             self.execute("MATCH (n) DETACH DELETE n")
+            return
+
+        logger.debug("No vertex_types provided and delete_all=False; no data deleted")
 
     def init_db(self, schema: Schema, recreate_schema: bool) -> None:
         """Initialize Neo4j with the given schema.
@@ -491,9 +499,12 @@ def init_db(self, schema: Schema, recreate_schema: bool) -> None:
     def clear_data(self, schema: Schema) -> None:
         """Remove all data from the graph without dropping the schema.
 
-        Deletes all nodes and relationships; labels (schema) remain.
+        Deletes nodes and relationships for schema-managed labels only.
         """
-        self.delete_graph_structure((), (), delete_all=True)
+        vc = schema.resolve_db_aware(DBType.NEO4J).vertex_config
+        vertex_types = tuple(vc.vertex_dbname(v) for v in vc.vertex_set)
+        if vertex_types:
+            self.delete_graph_structure(vertex_types=vertex_types)
 
     def upsert_docs_batch(self, docs, class_name, match_keys, **kwargs):
         """Upsert a batch of nodes using Cypher.
diff --git a/graflo/db/tigergraph/conn.py b/graflo/db/tigergraph/conn.py
index 77fd9915..f0324c89 100644
--- a/graflo/db/tigergraph/conn.py
+++ b/graflo/db/tigergraph/conn.py
@@ -2881,7 +2881,12 @@ def _parse_show_job_output(self, result_str: str) -> list[str]:
         """Parse SHOW JOB * output to extract job names."""
         return self._parse_show_output(result_str, "JOB")
 
-    def delete_graph_structure(self, vertex_types=(), graph_names=(), delete_all=False):
+    def delete_graph_structure(
+        self,
+        vertex_types: tuple[str, ...] | list[str] = (),
+        graph_names: tuple[str, ...] | list[str] = (),
+        delete_all: bool = False,
+    ) -> None:
         """
         Delete graph structure (graphs, vertex types, edge types) from TigerGraph.
 
@@ -2909,19 +2914,20 @@ def delete_graph_structure(self, vertex_types=(), graph_names=(), delete_all=Fal
                 # Step 1: Drop all graphs
                 graphs_to_drop = list(gnames) if gnames else []
 
-                # If no specific graphs provided, try to discover and drop all graphs
+                # Guardrail: never auto-discover and drop every graph by default.
+                # If no explicit graph target is provided, constrain to current graph.
                 if not graphs_to_drop:
-                    try:
-                        # Use GSQL to list all graphs
-                        show_graphs_cmd = "SHOW GRAPH *"
-                        result = self._execute_gsql(show_graphs_cmd)
-                        result_str = str(result)
-
-                        # Parse graph names using helper method
-                        graphs_to_drop = self._parse_show_graph_output(result_str)
-                    except Exception as e:
-                        logger.debug(f"Could not list graphs: {e}")
-                        graphs_to_drop = []
+                    current_graph = self.config.database
+                    if current_graph:
+                        graphs_to_drop = [current_graph]
+                        logger.warning(
+                            "delete_all=True without explicit graph_names: limiting TigerGraph teardown to current graph '%s'",
+                            current_graph,
+                        )
+                    else:
+                        raise ValueError(
+                            "Refusing global TigerGraph teardown without explicit graph_names or config.database"
+                        )
 
                 # Drop each graph
                 logger.info(
@@ -3100,9 +3106,11 @@ def clear_data(self, schema: Schema) -> None:
         Deletes vertices (and their edges) for all vertex types in the schema.
         """
         vc = schema.resolve_db_aware(DBType.TIGERGRAPH).vertex_config
+        graph_name = self.config.database or schema.metadata.name
         vertex_types = tuple(vc.vertex_dbname(v) for v in vc.vertex_set)
         if vertex_types:
-            self.delete_graph_structure(vertex_types=vertex_types)
+            with self._ensure_graph_context(graph_name=graph_name):
+                self.delete_graph_structure(vertex_types=vertex_types)
 
     def _generate_upsert_payload(
         self, data: list[dict[str, Any]], vname: str, vindex: tuple[str, ...]
diff --git a/graflo/hq/caster.py b/graflo/hq/caster.py
index 41eeac1f..ba68f0ff 100644
--- a/graflo/hq/caster.py
+++ b/graflo/hq/caster.py
@@ -20,10 +20,9 @@
 import sys
 import traceback
 from pathlib import Path
-from typing import Any, Literal, cast
+from typing import Any, cast
 
 import pandas as pd
-from pydantic import BaseModel, ConfigDict, Field
 
 from suthing import Timer
 
@@ -41,56 +40,41 @@
 from graflo.util.chunker import ChunkerType
 from graflo.architecture.contract.bindings import Bindings
 from graflo.hq.connection_provider import ConnectionProvider, EmptyConnectionProvider
+from graflo.hq.ingestion_parameters import (
+    CastBatchResult,
+    IngestionParams,
+    RowCastFailure,
+    RowErrorBudgetExceeded,
+)
 
 logger = logging.getLogger(__name__)
 
 _ROW_ERROR_TRACEBACK_MAX_CHARS = 16_384
 
 
-class RowErrorBudgetExceeded(RuntimeError):
-    """Raised when total row cast failures exceed ``IngestionParams.max_row_errors``."""
-
-    def __init__(
-        self,
-        *,
-        total_failures: int,
-        limit: int,
-        dead_letter_path: Path | None,
-    ) -> None:
-        self.total_failures = total_failures
-        self.limit = limit
-        self.dead_letter_path = dead_letter_path
-        dl = str(dead_letter_path) if dead_letter_path else "(not configured)"
-        super().__init__(
-            f"Row error budget exceeded: {total_failures} total failures "
-            f"(limit {limit}). Dead letter: {dl}"
-        )
-
-
-class RowCastFailure(BaseModel):
-    """Structured record for a single row that failed during resource casting."""
-
-    resource_name: str
-    row_index: int
-    exception_type: str
-    message: str
-    traceback: str = Field(
-        default="",
-        description="Formatted traceback, truncated to the configured max length.",
-    )
-    doc_preview: Any = Field(
-        default=None,
-        description="Subset or truncated JSON of the source document for debugging.",
-    )
+def _filter_graph_container_by_vertices_inplace(
+    gc: GraphContainer, *, allowed_vertex_names: set[str] | None
+) -> None:
+    """Restrict persistence to a subset of vertex types.
 
+    Mutates *gc* in-place, removing:
+    - vertex collections whose names are not in *allowed_vertex_names*
+    - edge collections whose source/target vertex names are not allowed
+    """
 
-class CastBatchResult(BaseModel):
-    """Outcome of casting a batch through a resource (possibly with skipped rows)."""
-
-    model_config = ConfigDict(arbitrary_types_allowed=True)
+    if allowed_vertex_names is None:
+        return
 
-    graph: GraphContainer
-    failures: list[RowCastFailure] = Field(default_factory=list)
+    gc.vertices = {
+        vcol: items
+        for vcol, items in gc.vertices.items()
+        if vcol in allowed_vertex_names
+    }
+    gc.edges = {
+        (vfrom, vto, rel): items
+        for (vfrom, vto, rel), items in gc.edges.items()
+        if vfrom in allowed_vertex_names and vto in allowed_vertex_names
+    }
 
 
 def _format_traceback(exc: BaseException) -> str:
@@ -136,72 +120,6 @@ def _row_failure_from_exception(
     )
 
 
-class IngestionParams(BaseModel):
-    """Parameters for controlling the ingestion process.
-
-    Attributes:
-        clear_data: If True, remove all existing graph data before ingestion without
-            changing the schema.
-        n_cores: Number of CPU cores/threads to use for parallel processing
-        max_items: Maximum number of items to process per resource (applies to all data sources)
-        batch_size: Size of batches for processing
-        dry: Whether to perform a dry run (no database changes)
-        init_only: Whether to only initialize the database without ingestion
-        limit_files: Optional limit on number of files to process
-        max_concurrent_db_ops: Maximum number of concurrent database operations (for vertices/edges).
-            If None, uses n_cores. Set to 1 to prevent deadlocks in databases that don't handle
-            concurrent transactions well (e.g., Neo4j). Database-independent setting.
-        datetime_after: Inclusive lower bound for datetime filtering (ISO format).
-            Rows with date_column >= datetime_after are included. Used with SQL/table sources.
-        datetime_before: Exclusive upper bound for datetime filtering (ISO format).
-            Rows with date_column < datetime_before are included. Range is [datetime_after, datetime_before).
-        datetime_column: Default column name for datetime filtering when the connector does not
-            specify date_field. Per-table override: set date_field on TableConnector (or FileConnector).
-        strict_references: If True, fail fast during model/resource initialization when
-            named references cannot be resolved (for example, a
-            ``transform.call.use`` value that does not exist in
-            ``ingestion_model.transforms``). If False, unresolved references may be
-            tolerated by legacy paths.
-        strict_registry: If True, fail registry build when resources cannot be wired to
-            concrete sources/connectors (missing connector/type/mismatch/source build
-            errors). If False, those issues are logged and skipped, allowing partial
-            ingestion.
-        dynamic_edges: If True, feedback edge declarations discovered during resource
-            runtime initialization (e.g. edge actors) into the shared schema edge
-            config. Keep False to preserve pure logical-schema immutability.
-        on_row_error: ``skip`` continues the batch on per-row cast errors (default);
-            ``fail`` fails the batch on the first error (legacy behavior).
-        row_error_dead_letter_path: If set, append one JSON line per failed row
-            (JSONL) for debugging.
-        max_row_errors: If set, total failed rows across the ingest run must not
-            exceed this value or :class:`RowErrorBudgetExceeded` is raised.
-        row_error_doc_preview_max_bytes: Max UTF-8 size for serialized ``doc_preview``.
-        row_error_doc_keys: If set, only these keys from the source doc appear in
-            ``doc_preview`` (recommended when documents may contain sensitive fields).
-    """
-
-    clear_data: bool = False
-    n_cores: int = 1
-    max_items: int | None = None
-    batch_size: int = 10000
-    dry: bool = False
-    init_only: bool = False
-    limit_files: int | None = None
-    max_concurrent_db_ops: int | None = None
-    datetime_after: str | None = None
-    datetime_before: str | None = None
-    datetime_column: str | None = None
-    # Strict contract checks for major-release style validation workflows.
-    strict_references: bool = True
-    strict_registry: bool = True
-    dynamic_edges: bool = False
-    on_row_error: Literal["skip", "fail"] = "skip"
-    row_error_dead_letter_path: Path | None = None
-    max_row_errors: int | None = None
-    row_error_doc_preview_max_bytes: int = 4096
-    row_error_doc_keys: tuple[str, ...] | None = None
-
-
 class Caster:
     """Main class for data casting and ingestion.
 
@@ -239,6 +157,7 @@ def __init__(
         self.ingestion_params = ingestion_params
         self.schema = schema
         self.ingestion_model = ingestion_model
+        self._allowed_vertex_names: set[str] | None = None
         self._row_error_total = 0
         self._row_error_io_lock = asyncio.Lock()
 
@@ -307,6 +226,9 @@ async def process_doc(doc: dict[str, Any]) -> Any:
             coros = [process_doc(doc) for doc in doc_list]
             docs = await asyncio.gather(*coros)
             graph = GraphContainer.from_docs_list(docs)
+            _filter_graph_container_by_vertices_inplace(
+                graph, allowed_vertex_names=self._allowed_vertex_names
+            )
             return CastBatchResult(graph=graph, failures=[])
 
         raw = await asyncio.gather(
@@ -340,6 +262,9 @@ async def process_doc(doc: dict[str, Any]) -> Any:
         await self._persist_row_failures(failures)
 
         graph = GraphContainer.from_docs_list(docs)
+        _filter_graph_container_by_vertices_inplace(
+            graph, allowed_vertex_names=self._allowed_vertex_names
+        )
         return CastBatchResult(graph=graph, failures=failures)
 
     # ------------------------------------------------------------------
@@ -540,6 +465,7 @@ async def ingest_data_sources(
         data_source_registry: DataSourceRegistry,
         conn_conf: DBConfig,
         ingestion_params: IngestionParams | None = None,
+        allowed_resource_names: set[str] | None = None,
     ):
         """Ingest data from data sources in a registry.
 
@@ -565,6 +491,11 @@ async def ingest_data_sources(
 
         tasks: list[AbstractDataSource] = []
         for resource_name in self.ingestion_model._resources.keys():
+            if (
+                allowed_resource_names is not None
+                and resource_name not in allowed_resource_names
+            ):
+                continue
             data_sources = data_source_registry.get_data_sources(resource_name)
             if data_sources:
                 logger.info(
@@ -622,10 +553,14 @@ def ingest(
         db_flavor = target_db_config.connection_type
         self.schema.db_profile.db_flavor = db_flavor
         self.schema.finish_init()
+
+        allowed_resource_names = self._resolve_ingestion_scope(ingestion_params)
+
         self.ingestion_model.finish_init(
             self.schema.core_schema,
             strict_references=ingestion_params.strict_references,
             dynamic_edge_feedback=ingestion_params.dynamic_edges,
+            allowed_vertex_names=self._allowed_vertex_names,
         )
 
         registry = RegistryBuilder(self.schema, self.ingestion_model).build(
@@ -640,6 +575,7 @@ def ingest(
                 data_source_registry=registry,
                 conn_conf=target_db_config,
                 ingestion_params=ingestion_params,
+                allowed_resource_names=allowed_resource_names,
             )
         )
 
@@ -647,6 +583,43 @@ def ingest(
     # Internal helpers
     # ------------------------------------------------------------------
 
+    def _resolve_ingestion_scope(
+        self, ingestion_params: IngestionParams
+    ) -> set[str] | None:
+        """Resolve and validate resource/vertex filters for ingestion.
+
+        Resolution order is resources first, then vertices.
+        """
+        if ingestion_params.resources is not None:
+            known_resources = set(self.ingestion_model._resources.keys())
+            requested_resources = set(ingestion_params.resources)
+            unknown_resources = requested_resources - known_resources
+            if unknown_resources:
+                raise ValueError(
+                    "Unknown resources in ingestion_params.resources: "
+                    + ", ".join(sorted(unknown_resources))
+                )
+            allowed_resource_names: set[str] | None = requested_resources
+        else:
+            allowed_resource_names = None
+
+        if ingestion_params.vertices is not None:
+            known_vertices = {
+                v.name for v in self.schema.core_schema.vertex_config.vertices
+            }
+            requested_vertices = set(ingestion_params.vertices)
+            unknown_vertices = requested_vertices - known_vertices
+            if unknown_vertices:
+                raise ValueError(
+                    "Unknown vertices in ingestion_params.vertices: "
+                    + ", ".join(sorted(unknown_vertices))
+                )
+            self._allowed_vertex_names = requested_vertices
+        else:
+            self._allowed_vertex_names = None
+
+        return allowed_resource_names
+
     def _make_db_writer(self) -> DBWriter:
         """Create a :class:`DBWriter` from the current ingestion params."""
         max_concurrent = (
diff --git a/graflo/hq/connection_provider.py b/graflo/hq/connection_provider.py
index e9df5c63..e4e35e53 100644
--- a/graflo/hq/connection_provider.py
+++ b/graflo/hq/connection_provider.py
@@ -123,13 +123,49 @@ def bind_connector_to_conn_proxy(
 
     def bind_from_bindings(self, *, bindings: Bindings) -> None:
         """Populate ``proxy_by_connector_hash`` from the contract bindings."""
-        for entry in bindings.connector_connection_bindings:
-            for connector in bindings.connectors:
-                if (
-                    entry.connector == connector.hash
-                    or entry.connector == connector.name
-                ):
-                    self.proxy_by_connector_hash[connector.hash] = entry.conn_proxy
+        for connector in bindings.connectors:
+            proxy = bindings.get_conn_proxy_for_connector(connector)
+            if proxy is not None:
+                self.proxy_by_connector_hash[connector.hash] = proxy
+
+    def bind_single_config_for_bindings(
+        self,
+        *,
+        bindings: Bindings,
+        conn_proxy: str,
+        config: GeneralizedConnConfig,
+    ) -> None:
+        """Bind one generalized config to all connectors in bindings.
+
+        This is intended for the common case where a single source DB
+        (or single generalized API endpoint) supplies all SQL/SPARQL connectors
+        in the manifest.
+
+        Raises:
+            ValueError: if bindings use multiple different ``conn_proxy`` labels.
+        """
+        used_proxies: set[str] = set()
+        for connector in bindings.connectors:
+            proxy = bindings.get_conn_proxy_for_connector(connector)
+            if proxy is not None:
+                used_proxies.add(proxy)
+
+        if not used_proxies:
+            raise ValueError(
+                "No connector_connection mappings found in bindings; "
+                "expected connector -> conn_proxy rows."
+            )
+
+        if used_proxies != {conn_proxy}:
+            used = ", ".join(sorted(used_proxies))
+            raise ValueError(
+                f"Expected all connector_connection mappings to use conn_proxy='{conn_proxy}', "
+                f"but found proxies: {used}. For multi-proxy setups, bind explicitly "
+                "with register_generalized_config(...) and bind_from_bindings(...)."
+            )
+
+        self.register_generalized_config(conn_proxy=conn_proxy, config=config)
+        self.bind_from_bindings(bindings=bindings)
 
     def get_generalized_conn_config(
         self, connector: ResourceConnector
diff --git a/graflo/hq/ingestion_parameters.py b/graflo/hq/ingestion_parameters.py
new file mode 100644
index 00000000..5a67d4f0
--- /dev/null
+++ b/graflo/hq/ingestion_parameters.py
@@ -0,0 +1,88 @@
+"""Ingestion parameters and row-error models for the caster.
+
+This module exists to keep `graflo/hq/caster.py` focused on casting logic, while
+keeping ingestion-policy types stable and importable.
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+from typing import Any, Literal
+
+from pydantic import BaseModel, ConfigDict, Field
+
+from graflo.architecture.graph_types import GraphContainer
+
+
+class RowErrorBudgetExceeded(RuntimeError):
+    """Raised when total row cast failures exceed ``IngestionParams.max_row_errors``."""
+
+    def __init__(
+        self,
+        *,
+        total_failures: int,
+        limit: int,
+        dead_letter_path: Path | None,
+    ) -> None:
+        self.total_failures = total_failures
+        self.limit = limit
+        self.dead_letter_path = dead_letter_path
+        dl = str(dead_letter_path) if dead_letter_path else "(not configured)"
+        super().__init__(
+            f"Row error budget exceeded: {total_failures} total failures "
+            f"(limit {limit}). Dead letter: {dl}"
+        )
+
+
+class RowCastFailure(BaseModel):
+    """Structured record for a single row that failed during resource casting."""
+
+    resource_name: str
+    row_index: int
+    exception_type: str
+    message: str
+    traceback: str = Field(
+        default="",
+        description="Formatted traceback, truncated to the configured max length.",
+    )
+    doc_preview: Any = Field(
+        default=None,
+        description="Subset or truncated JSON of the source document for debugging.",
+    )
+
+
+class CastBatchResult(BaseModel):
+    """Outcome of casting a batch through a resource (possibly with skipped rows)."""
+
+    model_config = ConfigDict(arbitrary_types_allowed=True)
+
+    graph: GraphContainer
+    failures: list[RowCastFailure] = Field(default_factory=list)
+
+
+class IngestionParams(BaseModel):
+    """Parameters for controlling the ingestion process."""
+
+    clear_data: bool = False
+    n_cores: int = 1
+    max_items: int | None = None
+    batch_size: int = 10000
+    dry: bool = False
+    init_only: bool = False
+    limit_files: int | None = None
+    resources: list[str] | None = None
+    vertices: list[str] | None = None
+    max_concurrent_db_ops: int | None = None
+    datetime_after: str | None = None
+    datetime_before: str | None = None
+    datetime_column: str | None = None
+
+    # Strict contract checks for major-release style validation workflows.
+    strict_references: bool = True
+    strict_registry: bool = True
+    dynamic_edges: bool = False
+    on_row_error: Literal["skip", "fail"] = "skip"
+    row_error_dead_letter_path: Path | None = None
+    max_row_errors: int | None = None
+    row_error_doc_preview_max_bytes: int = 4096
+    row_error_doc_keys: tuple[str, ...] | None = None
diff --git a/graflo/hq/registry_builder.py b/graflo/hq/registry_builder.py
index 4d4e6fbe..924f558f 100644
--- a/graflo/hq/registry_builder.py
+++ b/graflo/hq/registry_builder.py
@@ -68,8 +68,14 @@ def build(
         provider = connection_provider or EmptyConnectionProvider()
         failures: list[str] = []
 
+        resources_filter: set[str] | None = None
+        if ingestion_params.resources is not None:
+            resources_filter = set(ingestion_params.resources)
+
         for resource in self.ingestion_model.resources:
             resource_name = resource.name
+            if resources_filter is not None and resource_name not in resources_filter:
+                continue
             resource_type = bindings.get_resource_type(resource_name)
 
             if resource_type is None:
diff --git a/mkdocs.yml b/mkdocs.yml
index 8cfbe7d3..e09c258a 100644
--- a/mkdocs.yml
+++ b/mkdocs.yml
@@ -66,7 +66,9 @@ nav:
     -   4 - Dynamic Relations from Keys (Neo4j): examples/example-4.md
     -   5 - Ingesting Data from PostgreSQL Tables: examples/example-5.md
     -   6 - RDF / Turtle Ingestion with Explicit Resource Mapping: examples/example-6.md
-    -   7 - Multi-Edge Weights with Filters and dress Transform: examples/example-7.md
+    -   7 - Polymorphic Objects and Relations (vertex_router / edge_router): examples/example-7.md
+    -   8 - Multi-Edge Weights with Filters and dress Transform: examples/example-8.md
+    -   9 - Explicit connector_connection Proxy Wiring: examples/example-9.md
 -   API Reference: reference/
 -   Contributing: contributing.md
 markdown_extensions:
diff --git a/pyproject.toml b/pyproject.toml
index 3ce0f727..0596bca9 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -38,7 +38,7 @@ description = "A framework for transforming tabular (CSV, SQL) and hierarchical
 name = "graflo"
 readme = "README.md"
 requires-python = ">=3.11"
-version = "1.7.7"
+version = "1.7.8"
 
 [project.optional-dependencies]
 dev = [
diff --git a/test/architecture/test_connector_conn_proxy.py b/test/architecture/test_connector_conn_proxy.py
index ff757b6f..bf57330e 100644
--- a/test/architecture/test_connector_conn_proxy.py
+++ b/test/architecture/test_connector_conn_proxy.py
@@ -79,3 +79,147 @@ def test_provider_resolves_connector_based_config_for_multiple_resources() -> No
     assert cfg2 is not None
     assert cfg1.uri == pg_cfg.uri
     assert cfg2.uri == pg_cfg.uri
+
+
+def test_provider_bind_from_bindings_supports_resource_alias() -> None:
+    # connector.name omitted; manifests can use connector.resource_name as alias.
+    connector = TableConnector(
+        table_name="t1",
+        schema_name="public",
+        resource_name="people",
+    )
+    pg_cfg = PostgresConfig(
+        uri="postgresql://localhost:5432/db",
+        username="u",
+        password="p",
+        database="db",
+        schema_name="public",
+    )
+
+    bindings = Bindings(
+        connectors=[connector],
+        connector_connection=[{"connector": "people", "conn_proxy": "pg"}],
+    )
+
+    provider = InMemoryConnectionProvider()
+    provider.register_generalized_config(
+        conn_proxy="pg",
+        config=PostgresGeneralizedConnConfig(config=pg_cfg),
+    )
+    provider.bind_from_bindings(bindings=bindings)
+
+    cfg = provider.get_postgres_config(resource_name="r1", connector=connector)
+    assert cfg is not None
+    assert cfg.uri == pg_cfg.uri
+
+
+def test_provider_bind_from_bindings_resolves_by_connector_name() -> None:
+    connector = TableConnector(table_name="t1", schema_name="public", name="c1")
+    pg_cfg = PostgresConfig(
+        uri="postgresql://localhost:5432/db",
+        username="u",
+        password="p",
+        database="db",
+        schema_name="public",
+    )
+
+    bindings = Bindings(
+        connectors=[connector],
+        connector_connection=[{"connector": "c1", "conn_proxy": "pg"}],
+    )
+
+    provider = InMemoryConnectionProvider()
+    provider.register_generalized_config(
+        conn_proxy="pg",
+        config=PostgresGeneralizedConnConfig(config=pg_cfg),
+    )
+    provider.bind_from_bindings(bindings=bindings)
+
+    cfg = provider.get_postgres_config(resource_name="r1", connector=connector)
+    assert cfg is not None
+    assert cfg.uri == pg_cfg.uri
+
+
+def test_provider_bind_from_bindings_resolves_by_connector_hash() -> None:
+    connector = TableConnector(table_name="t1", schema_name="public", name="c1")
+    pg_cfg = PostgresConfig(
+        uri="postgresql://localhost:5432/db",
+        username="u",
+        password="p",
+        database="db",
+        schema_name="public",
+    )
+
+    bindings = Bindings(
+        connectors=[connector],
+        connector_connection=[{"connector": connector.hash, "conn_proxy": "pg"}],
+    )
+
+    provider = InMemoryConnectionProvider()
+    provider.register_generalized_config(
+        conn_proxy="pg",
+        config=PostgresGeneralizedConnConfig(config=pg_cfg),
+    )
+    provider.bind_from_bindings(bindings=bindings)
+
+    cfg = provider.get_postgres_config(resource_name="r1", connector=connector)
+    assert cfg is not None
+    assert cfg.uri == pg_cfg.uri
+
+
+def test_bind_single_config_for_bindings_binds_and_validates() -> None:
+    connector = TableConnector(
+        table_name="t1",
+        schema_name="public",
+        resource_name="people",
+    )
+    pg_cfg = PostgresConfig(
+        uri="postgresql://localhost:5432/db",
+        username="u",
+        password="p",
+        database="db",
+        schema_name="public",
+    )
+
+    bindings = Bindings(
+        connectors=[connector],
+        connector_connection=[{"connector": "people", "conn_proxy": "pg"}],
+    )
+
+    provider = InMemoryConnectionProvider()
+    provider.bind_single_config_for_bindings(
+        bindings=bindings,
+        conn_proxy="pg",
+        config=PostgresGeneralizedConnConfig(config=pg_cfg),
+    )
+
+    cfg = provider.get_postgres_config(resource_name="r1", connector=connector)
+    assert cfg is not None
+    assert cfg.uri == pg_cfg.uri
+
+    # Mismatch between helper conn_proxy and manifest conn_proxy should fail fast.
+    bindings_mismatch = Bindings(
+        connectors=[
+            TableConnector(
+                table_name="t1",
+                schema_name="public",
+                resource_name="people",
+            ),
+            TableConnector(
+                table_name="t2",
+                schema_name="public",
+                resource_name="products",
+            ),
+        ],
+        connector_connection=[
+            {"connector": "people", "conn_proxy": "pg"},
+            {"connector": "products", "conn_proxy": "pg2"},
+        ],
+    )
+    provider_mismatch = InMemoryConnectionProvider()
+    with pytest.raises(ValueError, match="Expected all connector_connection mappings"):
+        provider_mismatch.bind_single_config_for_bindings(
+            bindings=bindings_mismatch,
+            conn_proxy="pg",
+            config=PostgresGeneralizedConnConfig(config=pg_cfg),
+        )
diff --git a/test/hq/test_ingestion_subset.py b/test/hq/test_ingestion_subset.py
new file mode 100644
index 00000000..7750e1a4
--- /dev/null
+++ b/test/hq/test_ingestion_subset.py
@@ -0,0 +1,227 @@
+from __future__ import annotations
+
+from graflo.architecture.contract.declarations.resource import Resource
+from graflo.architecture.schema.edge import EdgeConfig
+from graflo.architecture.schema.vertex import VertexConfig
+from graflo.architecture.graph_types import GraphContainer
+from graflo.hq.caster import (
+    IngestionParams,
+    _filter_graph_container_by_vertices_inplace,
+)
+
+
+def test_filter_graph_container_by_vertices_keeps_allowed_vertices() -> None:
+    gc = GraphContainer(
+        vertices={
+            "A": [{"id": 1}],
+            "B": [{"id": 2}],
+            "C": [{"id": 3}],
+        },
+        edges={
+            ("A", "B", None): [({"id": 1}, {"id": 2}, {})],
+            ("A", "C", None): [({"id": 1}, {"id": 3}, {})],
+        },
+        linear=[],
+    )
+
+    _filter_graph_container_by_vertices_inplace(gc, allowed_vertex_names={"A", "B"})
+
+    assert set(gc.vertices.keys()) == {"A", "B"}
+    assert set(gc.edges.keys()) == {("A", "B", None)}
+
+
+def test_filter_graph_container_by_vertices_empty_ingests_nothing() -> None:
+    gc = GraphContainer(
+        vertices={
+            "A": [{"id": 1}],
+            "B": [{"id": 2}],
+        },
+        edges={
+            ("A", "B", None): [({"id": 1}, {"id": 2}, {})],
+        },
+        linear=[],
+    )
+
+    _filter_graph_container_by_vertices_inplace(gc, allowed_vertex_names=set())
+
+    assert gc.vertices == {}
+    assert gc.edges == {}
+
+
+def test_ingestion_params_accepts_resource_and_vertex_filters() -> None:
+    params = IngestionParams(resources=["r1"], vertices=["v1"])
+    assert params.resources == ["r1"]
+    assert params.vertices == ["v1"]
+
+
+def _vertex_config_a_b_c() -> VertexConfig:
+    return VertexConfig.from_dict(
+        {
+            "vertices": [
+                {"name": "A", "fields": ["id"], "identity": ["id"]},
+                {"name": "B", "fields": ["id"], "identity": ["id"]},
+                {"name": "C", "fields": ["id"], "identity": ["id"]},
+            ]
+        }
+    )
+
+
+def _edge_config_a_b_and_a_c() -> EdgeConfig:
+    return EdgeConfig.from_dict(
+        {"edges": [{"source": "A", "target": "B"}, {"source": "A", "target": "C"}]}
+    )
+
+
+def test_vertex_actor_early_exit_skips_disallowed_vertices() -> None:
+    vc = _vertex_config_a_b_c()
+    ec = EdgeConfig.from_dict({"edges": []})
+
+    resource = Resource.from_dict(
+        {
+            "name": "r",
+            "pipeline": [
+                {"vertex": "A", "from": {"id": "a_id"}},
+                {"vertex": "B", "from": {"id": "b_id"}},
+            ],
+        }
+    )
+    resource.finish_init(
+        vertex_config=vc,
+        edge_config=ec,
+        transforms={},
+        allowed_vertex_names={"A"},
+    )
+
+    extraction_ctx = resource._executor.extract({"a_id": "a1", "b_id": "b1"})
+    assert "A" in extraction_ctx.acc_vertex
+    assert "B" not in extraction_ctx.acc_vertex
+
+
+def test_vertex_router_early_exit_skips_disallowed_types() -> None:
+    vc = _vertex_config_a_b_c()
+    ec = EdgeConfig.from_dict({"edges": []})
+
+    resource = Resource.from_dict(
+        {
+            "name": "r",
+            "pipeline": [
+                {
+                    "vertex_router": {
+                        "type_field": "vtype",
+                        "vertex_from_map": {
+                            "A": {"id": "id"},
+                            "B": {"id": "id"},
+                        },
+                    }
+                }
+            ],
+        }
+    )
+
+    resource.finish_init(
+        vertex_config=vc,
+        edge_config=ec,
+        transforms={},
+        allowed_vertex_names={"A"},
+    )
+
+    extraction_ctx = resource._executor.extract({"vtype": "B", "id": "b1"})
+    assert "A" not in extraction_ctx.acc_vertex
+    assert "B" not in extraction_ctx.acc_vertex
+
+    extraction_ctx = resource._executor.extract({"vtype": "A", "id": "a1"})
+    assert "A" in extraction_ctx.acc_vertex
+    assert "B" not in extraction_ctx.acc_vertex
+
+
+def test_edge_router_early_exit_skips_disallowed_endpoints() -> None:
+    vc = _vertex_config_a_b_c()
+    ec = EdgeConfig.from_dict({"edges": []})
+
+    resource = Resource.from_dict(
+        {
+            "name": "r",
+            "pipeline": [
+                {
+                    "edge_router": {
+                        "source_type_field": "src_type",
+                        "target_type_field": "tgt_type",
+                        "source_fields": {"id": "src_id"},
+                        "target_fields": {"id": "tgt_id"},
+                    }
+                }
+            ],
+        }
+    )
+
+    resource.finish_init(
+        vertex_config=vc,
+        edge_config=ec,
+        transforms={},
+        allowed_vertex_names={"A", "B"},
+    )
+
+    extraction_ctx = resource._executor.extract(
+        {"src_type": "A", "src_id": "a1", "tgt_type": "C", "tgt_id": "c1"}
+    )
+    assert "A" not in extraction_ctx.acc_vertex  # no endpoint populated
+    assert "C" not in extraction_ctx.acc_vertex
+    assert not extraction_ctx.edge_intents
+
+    extraction_ctx = resource._executor.extract(
+        {"src_type": "A", "src_id": "a1", "tgt_type": "B", "tgt_id": "b1"}
+    )
+    assert "A" in extraction_ctx.acc_vertex
+    assert "B" in extraction_ctx.acc_vertex
+    assert extraction_ctx.edge_intents
+
+    result = resource._executor.assemble_result(extraction_ctx).entities
+    # No edge should involve disallowed vertex types.
+    for k in result.keys():
+        if not isinstance(k, tuple):
+            continue
+        s, t, _rel = k
+        assert s in {"A", "B"}
+        assert t in {"A", "B"}
+
+
+def test_edge_inference_skips_edges_with_disallowed_vertices() -> None:
+    vc = _vertex_config_a_b_c()
+    ec = _edge_config_a_b_and_a_c()
+
+    # Include an explicit A->C edge actor to ensure EdgeActor also early-exits.
+    resource = Resource.from_dict(
+        {
+            "name": "r",
+            "pipeline": [
+                {"vertex": "A", "from": {"id": "a_id"}},
+                {"vertex": "B", "from": {"id": "b_id"}},
+                {"vertex": "C", "from": {"id": "c_id"}},
+                {"edge": {"from": "A", "to": "C"}},
+            ],
+        }
+    )
+
+    resource.finish_init(
+        vertex_config=vc,
+        edge_config=ec,
+        transforms={},
+        allowed_vertex_names={"A", "B"},
+    )
+
+    extraction_ctx = resource._executor.extract(
+        {"a_id": "a1", "b_id": "b1", "c_id": "c1"}
+    )
+    assert "C" not in extraction_ctx.acc_vertex
+
+    result = resource._executor.assemble_result(extraction_ctx).entities
+    for k in result.keys():
+        if not isinstance(k, tuple):
+            continue
+        s, t, _rel = k
+        assert "C" not in (s, t)
+
+    # Edge A->B should still exist (inferred).
+    assert any(
+        isinstance(k, tuple) and k[0] == "A" and k[1] == "B" for k in result.keys()
+    )
diff --git a/uv.lock b/uv.lock
index f6d1ef47..870f24ee 100644
--- a/uv.lock
+++ b/uv.lock
@@ -396,7 +396,7 @@ dependencies = [
 ]
 name = "graflo"
 source = {editable = "."}
-version = "1.7.7"
+version = "1.7.8"
 
 [package.metadata]
 provides-extras = ["dev", "docs", "plot"]