diff --git a/README.md b/README.md
index c824edf..c5b5fe2 100644
--- a/README.md
+++ b/README.md
@@ -456,7 +456,9 @@ base_parameters:
 ```
 
 and the model name follows the same schema:
-`${var.catalog_name}.${resources.schemas.ml_workspace.name}.fraud_detection`.
+`${var.catalog_name}.${resources.schemas.ml_workspace.name}.${var.model_name}`
+(where `model_name` is the bare name, default `fraud_detection`). The notebooks prepend
+the catalog and `ml_schema` to it to build the three-level `uc_model_name`.
 
 > **Why reference the resource, not a plain variable?** Development mode prefixes the schema
 > *resource* but not a hand-written variable, so a variable like `${short_name}_fraud` would
diff --git a/databricks.yml b/databricks.yml
index ad0dc07..c63b540 100644
--- a/databricks.yml
+++ b/databricks.yml
@@ -15,8 +15,10 @@ variables:
     description: MLflow experiment for model training runs (per-user path).
     default: /Users/${workspace.current_user.userName}/${bundle.target}-mlops-workshop-fraud
   model_name:
-    description: Unity Catalog registered model name (in the per-user workspace schema).
-    default: ${var.catalog_name}.${var.ml_schema}.fraud_detection
+    description: >-
+      Bare (unqualified) registered model name. Notebooks prepend catalog_name + ml_schema
+      to form the three-level Unity Catalog name (uc_model_name).
+    default: fraud_detection
   catalog_name:
     description: Unity Catalog catalog for the environment (dev/staging/prod).
     default: adoption_workshop
diff --git a/resources/deployment-job-workflow.yml b/resources/deployment-job-workflow.yml
index 4325023..a4ebea6 100644
--- a/resources/deployment-job-workflow.yml
+++ b/resources/deployment-job-workflow.yml
@@ -24,7 +24,7 @@ resources:
         enabled: true
       parameters:
         - name: model_name
-          default: ${var.model_name}
+          default: ${var.catalog_name}.${var.ml_schema}.${var.model_name}
         - name: model_version
           default: ""
       tasks:
@@ -35,6 +35,7 @@ resources:
             notebook_path: ../${var.code_root}/deployment/model_deployment/notebooks/evaluation.py
             base_parameters:
               catalog_name: ${var.catalog_name}
+              ml_schema: ${var.ml_schema}
               metric: roc_auc
               baseline: "0.65"
 
diff --git a/solution/deployment/batch_inference/notebooks/batch_inference.py b/solution/deployment/batch_inference/notebooks/batch_inference.py
index 37f4237..699c7ca 100644
--- a/solution/deployment/batch_inference/notebooks/batch_inference.py
+++ b/solution/deployment/batch_inference/notebooks/batch_inference.py
@@ -26,7 +26,7 @@
 dbutils.widgets.text("catalog_name", "adoption_workshop")
 dbutils.widgets.text("gold_schema", "fraud_gold")
 dbutils.widgets.text("ml_schema", "")
-dbutils.widgets.text("model_name", "")
+dbutils.widgets.text("model_name", "fraud_detection")
 dbutils.widgets.text("model_alias", "champion")
 
 catalog_name = dbutils.widgets.get("catalog_name")
@@ -43,14 +43,17 @@
     ml_schema = f"dev_{_short}_fraud"
 
 model_alias = dbutils.widgets.get("model_alias")
-# The batch-inference job passes the full three-level model name; fall back for interactive runs.
-model_name = dbutils.widgets.get("model_name") or f"{catalog_name}.{ml_schema}.fraud_detection"
+model_name = dbutils.widgets.get("model_name")
+# Accept either a bare name (prepend catalog + ml_schema) or an already-qualified 3-level name.
+uc_model_name = (
+    model_name if model_name.count(".") == 2 else f"{catalog_name}.{ml_schema}.{model_name}"
+)
 
 # Read the transactions to score from the shared gold source; write predictions to the personal schema.
 source_table = f"{catalog_name}.{gold_schema}.transactions_enriched"
 predictions_table = f"{catalog_name}.{ml_schema}.fraud_predictions"
 
-print(f"Model:       {model_name}@{model_alias}")
+print(f"Model:       {uc_model_name}@{model_alias}")
 print(f"Score from:  {source_table}")
 print(f"Write to:    {predictions_table}")
 
@@ -77,7 +80,7 @@
 # Resolve the champion alias to a concrete version, to stamp every scored row with the model
 # version that produced it (the monitor's model_id_col groups metrics by this).
 client = MlflowClient()
-champion = client.get_model_version_by_alias(model_name, model_alias)
+champion = client.get_model_version_by_alias(uc_model_name, model_alias)
 model_version = str(champion.version)
 print(f"{model_alias} -> version {model_version}")
 
@@ -121,10 +124,10 @@
 # COMMAND ----------
 
 # TODO-BEGIN: score the batch with the champion model
-# HINT: fe.score_batch(model_uri=f"models:/{model_name}@{model_alias}", df=to_score).
+# HINT: fe.score_batch(model_uri=f"models:/{uc_model_name}@{model_alias}", df=to_score).
 # HINT: the result has the lookup key plus a "prediction" column (the monitor's prediction_col).
 scored = fe.score_batch(
-    model_uri=f"models:/{model_name}@{model_alias}",
+    model_uri=f"models:/{uc_model_name}@{model_alias}",
     df=to_score,
 )
 # TODO-END
diff --git a/solution/deployment/model_deployment/notebooks/approval.py b/solution/deployment/model_deployment/notebooks/approval.py
index 7c518af..6346aff 100644
--- a/solution/deployment/model_deployment/notebooks/approval.py
+++ b/solution/deployment/model_deployment/notebooks/approval.py
@@ -20,22 +20,59 @@
 
 # COMMAND ----------
 
+# The deployment job injects the FULL three-level model name (catalog.schema.model) into
+# `model_name`, plus the `model_version` it is gating. Both are empty when this notebook is
+# run interactively, so the fallback below rebuilds them.
 dbutils.widgets.text("model_name", "")
 dbutils.widgets.text("model_version", "")
+dbutils.widgets.text("catalog_name", "adoption_workshop")
+dbutils.widgets.text("ml_schema", "")
 dbutils.widgets.text("approval_tag_name", "")
 dbutils.widgets.text("auto_approve", "false")
 
-model_name = dbutils.widgets.get("model_name")
-model_version = dbutils.widgets.get("model_version")
+uc_model_name = dbutils.widgets.get("model_name").strip()
+model_version = dbutils.widgets.get("model_version").strip()
+catalog_name = dbutils.widgets.get("catalog_name")
+ml_schema = dbutils.widgets.get("ml_schema")
 # The deployment job passes the task name here (e.g. "Approval_Check") as the tag key.
 tag_name = dbutils.widgets.get("approval_tag_name") or "Approval_Check"
 auto_approve = dbutils.widgets.get("auto_approve").strip().lower() == "true"
 
-assert model_name and model_version, (
-    "model_name and model_version are injected by the deployment job."
-)
+# Fail fast on a half-configured deployment job: model_name and model_version are injected
+# together. Both present = job; both empty = interactive run (derived below). Exactly one
+# present is a misconfiguration that would silently gate the wrong model or version.
+if bool(uc_model_name) != bool(model_version):
+    raise ValueError(
+        "Only one of model_name / model_version was provided. Pass both (deployment job) "
+        "or neither (interactive run)."
+    )
+
+from pyspark.sql import functions as F
+
+# Interactive fallback: rebuild the personal schema and the full model name, and default to
+# the latest version, so the notebook can be run by hand (outside the deployment job).
+if not ml_schema:
+    _user = spark.range(1).select(F.current_user()).first()[0]
+    _short = "".join(c if c.isalnum() else "_" for c in _user.split("@")[0])
+    ml_schema = f"dev_{_short}_fraud"
+if not uc_model_name:
+    uc_model_name = f"{catalog_name}.{ml_schema}.fraud_detection"
+if not model_version:
+    import mlflow
+    from mlflow.tracking import MlflowClient
+
+    mlflow.set_registry_uri("databricks-uc")
+    _versions = [
+        int(mv.version)
+        for mv in MlflowClient().search_model_versions(f"name='{uc_model_name}'")
+    ]
+    if not _versions:
+        raise ValueError(
+            f"No model versions found for {uc_model_name!r}. Provide model_version explicitly."
+        )
+    model_version = str(max(_versions))
 print(
-    f"Approval check for {model_name} v{model_version} | tag='{tag_name}' | auto_approve={auto_approve}"
+    f"Approval check for {uc_model_name} v{model_version} | tag='{tag_name}' | auto_approve={auto_approve}"
 )
 
 # COMMAND ----------
@@ -47,10 +84,10 @@
 # In dev, auto-approve: set the tag so this run (and the UI) reflect an approved state, then
 # pass. In staging/prod, a human must have set the tag to "Approved".
 if auto_approve:
-    client.set_model_version_tag(model_name, model_version, tag_name, "Approved")
+    client.set_model_version_tag(uc_model_name, model_version, tag_name, "Approved")
     print(f"Auto-approved (dev): set {tag_name}=Approved on v{model_version}.")
 else:
-    mv = client.get_model_version(model_name, model_version)
+    mv = client.get_model_version(uc_model_name, model_version)
     status = (mv.tags or {}).get(tag_name)
     print(f"Current approval tag {tag_name}={status!r}")
     if status != "Approved":
diff --git a/solution/deployment/model_deployment/notebooks/deployment.py b/solution/deployment/model_deployment/notebooks/deployment.py
index 12c7de0..0eed33c 100644
--- a/solution/deployment/model_deployment/notebooks/deployment.py
+++ b/solution/deployment/model_deployment/notebooks/deployment.py
@@ -18,35 +18,59 @@
 
 # COMMAND ----------
 
-# The deployment job injects model_name + model_version as JOB-level parameters.
+# The deployment job injects the FULL three-level model name (catalog.schema.model) into
+# `model_name`, plus the `model_version` that triggered it. Both are empty when this notebook
+# is run interactively, so the fallback below rebuilds them.
 dbutils.widgets.text("model_name", "")
 dbutils.widgets.text("model_version", "")
 dbutils.widgets.text("catalog_name", "adoption_workshop")
 dbutils.widgets.text("ml_schema", "")
 
-model_name = dbutils.widgets.get("model_name")
-model_version = dbutils.widgets.get("model_version")
+uc_model_name = dbutils.widgets.get("model_name").strip()
+model_version = dbutils.widgets.get("model_version").strip()
 catalog_name = dbutils.widgets.get("catalog_name")
 ml_schema = dbutils.widgets.get("ml_schema")
-assert model_name and model_version, (
-    "model_name and model_version are injected by the deployment job."
-)
+
+# Fail fast on a half-configured deployment job: model_name and model_version are injected
+# together. Both present = job; both empty = interactive run (derived below). Exactly one
+# present is a misconfiguration that would silently gate the wrong model or version.
+if bool(uc_model_name) != bool(model_version):
+    raise ValueError(
+        "Only one of model_name / model_version was provided. Pass both (deployment job) "
+        "or neither (interactive run)."
+    )
 
 from pyspark.sql import functions as F
 
-# Interactive fallback: jobs pass the resolved personal schema; running standalone derives
-# the same per-user name the bundle uses (dev_<short>_fraud) so runs stay isolated.
+# Interactive fallback: rebuild the personal schema and the full model name, and default to
+# the latest version, so the notebook can be run by hand (outside the deployment job).
 if not ml_schema:
     _user = spark.range(1).select(F.current_user()).first()[0]
     _short = "".join(c if c.isalnum() else "_" for c in _user.split("@")[0])
     ml_schema = f"dev_{_short}_fraud"
+if not uc_model_name:
+    uc_model_name = f"{catalog_name}.{ml_schema}.fraud_detection"
+if not model_version:
+    import mlflow
+    from mlflow.tracking import MlflowClient
+
+    mlflow.set_registry_uri("databricks-uc")
+    _versions = [
+        int(mv.version)
+        for mv in MlflowClient().search_model_versions(f"name='{uc_model_name}'")
+    ]
+    if not _versions:
+        raise ValueError(
+            f"No model versions found for {uc_model_name!r}. Provide model_version explicitly."
+        )
+    model_version = str(max(_versions))
 
 endpoint_name = f"{ml_schema}_fraud"
 online_store_name = "fraud-workshop-online"
 card_feature_table = f"{catalog_name}.{ml_schema}.card_features"
 client_feature_table = f"{catalog_name}.{ml_schema}.client_features"
 
-print(f"Deploying {model_name} v{model_version} -> endpoint {endpoint_name}")
+print(f"Deploying {uc_model_name} v{model_version} -> endpoint {endpoint_name}")
 
 # COMMAND ----------
 
@@ -63,9 +87,9 @@
 mlflow.set_registry_uri("databricks-uc")
 client = MlflowClient()
 
-client.set_registered_model_alias(model_name, "champion", model_version)
+client.set_registered_model_alias(uc_model_name, "champion", model_version)
 try:
-    client.delete_registered_model_alias(model_name, "challenger")
+    client.delete_registered_model_alias(uc_model_name, "challenger")
 except Exception:
     pass  # there may be no challenger alias (e.g. first deployment)
 print(f"@champion -> version {model_version}")
@@ -131,7 +155,7 @@ def wait_online_store(name, minutes=20):
 
 served_entities = [
     ServedEntityInput(
-        entity_name=model_name,
+        entity_name=uc_model_name,
         entity_version=model_version,
         scale_to_zero_enabled=True,
         workload_size="Small",
diff --git a/solution/deployment/model_deployment/notebooks/evaluation.py b/solution/deployment/model_deployment/notebooks/evaluation.py
index 9d2003c..493150f 100644
--- a/solution/deployment/model_deployment/notebooks/evaluation.py
+++ b/solution/deployment/model_deployment/notebooks/evaluation.py
@@ -21,23 +21,58 @@
 
 # COMMAND ----------
 
-# The deployment job injects model_name + model_version as JOB-level parameters.
+# The deployment job injects the FULL three-level model name (catalog.schema.model) into
+# `model_name`, plus the `model_version` that triggered it. Both are empty when this notebook
+# is run interactively, so the fallback below rebuilds them.
 dbutils.widgets.text("model_name", "")
 dbutils.widgets.text("model_version", "")
 dbutils.widgets.text("catalog_name", "adoption_workshop")
+dbutils.widgets.text("ml_schema", "")
 dbutils.widgets.text("metric", "roc_auc")
 dbutils.widgets.text("baseline", "0.65")
 
-model_name = dbutils.widgets.get("model_name")
-model_version = dbutils.widgets.get("model_version")
+uc_model_name = dbutils.widgets.get("model_name").strip()
+model_version = dbutils.widgets.get("model_version").strip()
 catalog_name = dbutils.widgets.get("catalog_name")
+ml_schema = dbutils.widgets.get("ml_schema")
 metric = dbutils.widgets.get("metric")
 baseline = float(dbutils.widgets.get("baseline"))
 
-assert model_name and model_version, (
-    "model_name and model_version are injected by the deployment job."
-)
-print(f"Evaluating {model_name} version {model_version} on '{metric}' (floor {baseline}).")
+# Fail fast on a half-configured deployment job: model_name and model_version are injected
+# together. Both present = job; both empty = interactive run (derived below). Exactly one
+# present is a misconfiguration that would silently gate the wrong model or version.
+if bool(uc_model_name) != bool(model_version):
+    raise ValueError(
+        "Only one of model_name / model_version was provided. Pass both (deployment job) "
+        "or neither (interactive run)."
+    )
+
+from pyspark.sql import functions as F
+
+# Interactive fallback: rebuild the personal schema and the full model name the same way the
+# training/batch notebooks do, and default to the latest version, so this runs by hand too.
+if not ml_schema:
+    _user = spark.range(1).select(F.current_user()).first()[0]
+    _short = "".join(c if c.isalnum() else "_" for c in _user.split("@")[0])
+    ml_schema = f"dev_{_short}_fraud"
+if not uc_model_name:
+    uc_model_name = f"{catalog_name}.{ml_schema}.fraud_detection"
+if not model_version:
+    import mlflow
+    from mlflow.tracking import MlflowClient
+
+    mlflow.set_registry_uri("databricks-uc")
+    _versions = [
+        int(mv.version)
+        for mv in MlflowClient().search_model_versions(f"name='{uc_model_name}'")
+    ]
+    if not _versions:
+        raise ValueError(
+            f"No model versions found for {uc_model_name!r}. Provide model_version explicitly."
+        )
+    model_version = str(max(_versions))
+
+print(f"Evaluating {uc_model_name} version {model_version} on '{metric}' (floor {baseline}).")
 
 # COMMAND ----------
 
@@ -79,7 +114,7 @@
 
 def score_holdout(version):
     """Score a model version on the holdout; return a pandas DF of is_fraud + fraud score."""
-    scored = fe.score_batch(model_uri=f"models:/{model_name}/{version}", df=eval_spine)
+    scored = fe.score_batch(model_uri=f"models:/{uc_model_name}/{version}", df=eval_spine)
     return scored.select("is_fraud", F.col("prediction").cast("double").alias("score")).toPandas()
 
 
@@ -89,7 +124,7 @@ def score_holdout(version):
 # Fair comparison: re-score the current champion on the SAME holdout. If there is no champion
 # yet (first model), gate against the metric floor instead.
 try:
-    champion = client.get_model_version_by_alias(model_name, "champion")
+    champion = client.get_model_version_by_alias(uc_model_name, "champion")
     champ_pdf = score_holdout(champion.version)
     bar = float(roc_auc_score(champ_pdf["is_fraud"], champ_pdf["score"]))
     bar_label = f"champion (v{champion.version})"
@@ -99,9 +134,9 @@ def score_holdout(version):
 print(f"candidate {metric}={candidate_score:.4f}  vs  {bar_label}={bar:.4f}")
 
 # Surface the decision inputs on the model version so the approver sees them in the UI.
-client.set_model_version_tag(model_name, model_version, "eval_metric", metric)
-client.set_model_version_tag(model_name, model_version, "eval_score", f"{candidate_score:.4f}")
-client.set_model_version_tag(model_name, model_version, "eval_bar", f"{bar:.4f}")
+client.set_model_version_tag(uc_model_name, model_version, "eval_metric", metric)
+client.set_model_version_tag(uc_model_name, model_version, "eval_score", f"{candidate_score:.4f}")
+client.set_model_version_tag(uc_model_name, model_version, "eval_bar", f"{bar:.4f}")
 
 # COMMAND ----------
 
@@ -125,7 +160,7 @@ def score_holdout(version):
 ConfusionMatrixDisplay(cm, display_labels=["legit", "fraud"]).plot(
     cmap="Blues", ax=ax, colorbar=False
 )
-ax.set_title(f"{model_name.split('.')[-1]} v{model_version} (holdout, threshold 0.5)")
+ax.set_title(f"{uc_model_name.split('.')[-1]} v{model_version} (holdout, threshold 0.5)")
 fig.tight_layout()
 with mlflow.start_run(run_name=f"evaluation_v{model_version}") as ev_run:
     mlflow.log_figure(fig, "confusion_matrix.png")
diff --git a/solution/training/notebooks/training.py b/solution/training/notebooks/training.py
index 1a55220..b97d89a 100644
--- a/solution/training/notebooks/training.py
+++ b/solution/training/notebooks/training.py
@@ -25,7 +25,7 @@
 dbutils.widgets.text("gold_schema", "fraud_gold")
 dbutils.widgets.text("ml_schema", "")
 dbutils.widgets.text("experiment_name", "")
-dbutils.widgets.text("model_name", "")
+dbutils.widgets.text("model_name", "fraud_detection")
 dbutils.widgets.text("deployment_job_id", "")
 
 catalog_name = dbutils.widgets.get("catalog_name")
@@ -52,13 +52,17 @@
 experiment_name = (
     dbutils.widgets.get("experiment_name") or f"/Users/{current_user}/mlops-workshop-fraud"
 )
-model_name = dbutils.widgets.get("model_name") or f"{catalog_name}.{ml_schema}.fraud_detection"
+model_name = dbutils.widgets.get("model_name")
+# Accept either a bare name (prepend catalog + ml_schema) or an already-qualified 3-level name.
+uc_model_name = (
+    model_name if model_name.count(".") == 2 else f"{catalog_name}.{ml_schema}.{model_name}"
+)
 
 print(f"Card features:   {card_feature_table}")
 print(f"Client features: {client_feature_table}")
 print(f"Labels:          {source_table}")
 print(f"Experiment: {experiment_name}")
-print(f"Model:      {model_name}")
+print(f"Model:      {uc_model_name}")
 
 # COMMAND ----------
 
@@ -110,11 +114,11 @@
 
     _client = MlflowClient()
     try:
-        _client.get_registered_model(model_name)
-        _client.update_registered_model(model_name, deployment_job_id=deployment_job_id)
+        _client.get_registered_model(uc_model_name)
+        _client.update_registered_model(uc_model_name, deployment_job_id=deployment_job_id)
     except RestException:
-        _client.create_registered_model(model_name, deployment_job_id=deployment_job_id)
-    print(f"Connected {model_name} to deployment job {deployment_job_id}.")
+        _client.create_registered_model(uc_model_name, deployment_job_id=deployment_job_id)
+    print(f"Connected {uc_model_name} to deployment job {deployment_job_id}.")
 else:
     print("No deployment_job_id passed; skipping deployment-job connection (interactive run).")
 
@@ -318,13 +322,13 @@ def predict(self, context, model_input):
     # HINT:   reproduces the same features) and registers it to UC in a single call.
     # HINT: fe.log_model(model=FraudProbabilityModel(model), artifact_path="model",
     # HINT:   flavor=mlflow.pyfunc, training_set=training_set,
-    # HINT:   registered_model_name=model_name, infer_input_example=True)
+    # HINT:   registered_model_name=uc_model_name, infer_input_example=True)
     fe.log_model(
         model=FraudProbabilityModel(model),
         artifact_path="model",
         flavor=mlflow.pyfunc,
         training_set=training_set,
-        registered_model_name=model_name,
+        registered_model_name=uc_model_name,
         infer_input_example=True,
     )
     # TODO-END
@@ -416,10 +420,14 @@ def predict(self, context, model_input):
 client = MlflowClient()
 # Unity Catalog's search_model_versions only supports a `name='...'` filter (no run_id
 # filtering), so fetch this model's versions and take the highest: the one registered above.
-versions = client.search_model_versions(f"name='{model_name}'")
-new_version = max(int(mv.version) for mv in versions)
-client.set_registered_model_alias(model_name, "challenger", new_version)
-print(f"Registered {model_name} version {new_version} and set alias @challenger.")
+_versions = [int(mv.version) for mv in client.search_model_versions(f"name='{uc_model_name}'")]
+if not _versions:
+    raise ValueError(
+        f"No model versions found for {uc_model_name!r}. Was the model registered above?"
+    )
+new_version = max(_versions)
+client.set_registered_model_alias(uc_model_name, "challenger", new_version)
+print(f"Registered {uc_model_name} version {new_version} and set alias @challenger.")
 
 # COMMAND ----------
 
diff --git a/src/deployment/batch_inference/notebooks/batch_inference.py b/src/deployment/batch_inference/notebooks/batch_inference.py
index 1dc8932..53f2309 100644
--- a/src/deployment/batch_inference/notebooks/batch_inference.py
+++ b/src/deployment/batch_inference/notebooks/batch_inference.py
@@ -26,7 +26,7 @@
 dbutils.widgets.text("catalog_name", "adoption_workshop")
 dbutils.widgets.text("gold_schema", "fraud_gold")
 dbutils.widgets.text("ml_schema", "")
-dbutils.widgets.text("model_name", "")
+dbutils.widgets.text("model_name", "fraud_detection")
 dbutils.widgets.text("model_alias", "champion")
 
 catalog_name = dbutils.widgets.get("catalog_name")
@@ -43,14 +43,17 @@
     ml_schema = f"dev_{_short}_fraud"
 
 model_alias = dbutils.widgets.get("model_alias")
-# The batch-inference job passes the full three-level model name; fall back for interactive runs.
-model_name = dbutils.widgets.get("model_name") or f"{catalog_name}.{ml_schema}.fraud_detection"
+model_name = dbutils.widgets.get("model_name")
+# Accept either a bare name (prepend catalog + ml_schema) or an already-qualified 3-level name.
+uc_model_name = (
+    model_name if model_name.count(".") == 2 else f"{catalog_name}.{ml_schema}.{model_name}"
+)
 
 # Read the transactions to score from the shared gold source; write predictions to the personal schema.
 source_table = f"{catalog_name}.{gold_schema}.transactions_enriched"
 predictions_table = f"{catalog_name}.{ml_schema}.fraud_predictions"
 
-print(f"Model:       {model_name}@{model_alias}")
+print(f"Model:       {uc_model_name}@{model_alias}")
 print(f"Score from:  {source_table}")
 print(f"Write to:    {predictions_table}")
 
@@ -77,7 +80,7 @@
 # Resolve the champion alias to a concrete version, to stamp every scored row with the model
 # version that produced it (the monitor's model_id_col groups metrics by this).
 client = MlflowClient()
-champion = client.get_model_version_by_alias(model_name, model_alias)
+champion = client.get_model_version_by_alias(uc_model_name, model_alias)
 model_version = str(champion.version)
 print(f"{model_alias} -> version {model_version}")
 
@@ -121,7 +124,7 @@
 # COMMAND ----------
 
 # TODO: score the batch with the champion model
-# HINT: fe.score_batch(model_uri=f"models:/{model_name}@{model_alias}", df=to_score).
+# HINT: fe.score_batch(model_uri=f"models:/{uc_model_name}@{model_alias}", df=to_score).
 # HINT: the result has the lookup key plus a "prediction" column (the monitor's prediction_col).
 # <-- Your code here
 
diff --git a/src/deployment/model_deployment/notebooks/approval.py b/src/deployment/model_deployment/notebooks/approval.py
index 7c518af..6346aff 100644
--- a/src/deployment/model_deployment/notebooks/approval.py
+++ b/src/deployment/model_deployment/notebooks/approval.py
@@ -20,22 +20,59 @@
 
 # COMMAND ----------
 
+# The deployment job injects the FULL three-level model name (catalog.schema.model) into
+# `model_name`, plus the `model_version` it is gating. Both are empty when this notebook is
+# run interactively, so the fallback below rebuilds them.
 dbutils.widgets.text("model_name", "")
 dbutils.widgets.text("model_version", "")
+dbutils.widgets.text("catalog_name", "adoption_workshop")
+dbutils.widgets.text("ml_schema", "")
 dbutils.widgets.text("approval_tag_name", "")
 dbutils.widgets.text("auto_approve", "false")
 
-model_name = dbutils.widgets.get("model_name")
-model_version = dbutils.widgets.get("model_version")
+uc_model_name = dbutils.widgets.get("model_name").strip()
+model_version = dbutils.widgets.get("model_version").strip()
+catalog_name = dbutils.widgets.get("catalog_name")
+ml_schema = dbutils.widgets.get("ml_schema")
 # The deployment job passes the task name here (e.g. "Approval_Check") as the tag key.
 tag_name = dbutils.widgets.get("approval_tag_name") or "Approval_Check"
 auto_approve = dbutils.widgets.get("auto_approve").strip().lower() == "true"
 
-assert model_name and model_version, (
-    "model_name and model_version are injected by the deployment job."
-)
+# Fail fast on a half-configured deployment job: model_name and model_version are injected
+# together. Both present = job; both empty = interactive run (derived below). Exactly one
+# present is a misconfiguration that would silently gate the wrong model or version.
+if bool(uc_model_name) != bool(model_version):
+    raise ValueError(
+        "Only one of model_name / model_version was provided. Pass both (deployment job) "
+        "or neither (interactive run)."
+    )
+
+from pyspark.sql import functions as F
+
+# Interactive fallback: rebuild the personal schema and the full model name, and default to
+# the latest version, so the notebook can be run by hand (outside the deployment job).
+if not ml_schema:
+    _user = spark.range(1).select(F.current_user()).first()[0]
+    _short = "".join(c if c.isalnum() else "_" for c in _user.split("@")[0])
+    ml_schema = f"dev_{_short}_fraud"
+if not uc_model_name:
+    uc_model_name = f"{catalog_name}.{ml_schema}.fraud_detection"
+if not model_version:
+    import mlflow
+    from mlflow.tracking import MlflowClient
+
+    mlflow.set_registry_uri("databricks-uc")
+    _versions = [
+        int(mv.version)
+        for mv in MlflowClient().search_model_versions(f"name='{uc_model_name}'")
+    ]
+    if not _versions:
+        raise ValueError(
+            f"No model versions found for {uc_model_name!r}. Provide model_version explicitly."
+        )
+    model_version = str(max(_versions))
 print(
-    f"Approval check for {model_name} v{model_version} | tag='{tag_name}' | auto_approve={auto_approve}"
+    f"Approval check for {uc_model_name} v{model_version} | tag='{tag_name}' | auto_approve={auto_approve}"
 )
 
 # COMMAND ----------
@@ -47,10 +84,10 @@
 # In dev, auto-approve: set the tag so this run (and the UI) reflect an approved state, then
 # pass. In staging/prod, a human must have set the tag to "Approved".
 if auto_approve:
-    client.set_model_version_tag(model_name, model_version, tag_name, "Approved")
+    client.set_model_version_tag(uc_model_name, model_version, tag_name, "Approved")
     print(f"Auto-approved (dev): set {tag_name}=Approved on v{model_version}.")
 else:
-    mv = client.get_model_version(model_name, model_version)
+    mv = client.get_model_version(uc_model_name, model_version)
     status = (mv.tags or {}).get(tag_name)
     print(f"Current approval tag {tag_name}={status!r}")
     if status != "Approved":
diff --git a/src/deployment/model_deployment/notebooks/deployment.py b/src/deployment/model_deployment/notebooks/deployment.py
index 12c7de0..0eed33c 100644
--- a/src/deployment/model_deployment/notebooks/deployment.py
+++ b/src/deployment/model_deployment/notebooks/deployment.py
@@ -18,35 +18,59 @@
 
 # COMMAND ----------
 
-# The deployment job injects model_name + model_version as JOB-level parameters.
+# The deployment job injects the FULL three-level model name (catalog.schema.model) into
+# `model_name`, plus the `model_version` that triggered it. Both are empty when this notebook
+# is run interactively, so the fallback below rebuilds them.
 dbutils.widgets.text("model_name", "")
 dbutils.widgets.text("model_version", "")
 dbutils.widgets.text("catalog_name", "adoption_workshop")
 dbutils.widgets.text("ml_schema", "")
 
-model_name = dbutils.widgets.get("model_name")
-model_version = dbutils.widgets.get("model_version")
+uc_model_name = dbutils.widgets.get("model_name").strip()
+model_version = dbutils.widgets.get("model_version").strip()
 catalog_name = dbutils.widgets.get("catalog_name")
 ml_schema = dbutils.widgets.get("ml_schema")
-assert model_name and model_version, (
-    "model_name and model_version are injected by the deployment job."
-)
+
+# Fail fast on a half-configured deployment job: model_name and model_version are injected
+# together. Both present = job; both empty = interactive run (derived below). Exactly one
+# present is a misconfiguration that would silently gate the wrong model or version.
+if bool(uc_model_name) != bool(model_version):
+    raise ValueError(
+        "Only one of model_name / model_version was provided. Pass both (deployment job) "
+        "or neither (interactive run)."
+    )
 
 from pyspark.sql import functions as F
 
-# Interactive fallback: jobs pass the resolved personal schema; running standalone derives
-# the same per-user name the bundle uses (dev_<short>_fraud) so runs stay isolated.
+# Interactive fallback: rebuild the personal schema and the full model name, and default to
+# the latest version, so the notebook can be run by hand (outside the deployment job).
 if not ml_schema:
     _user = spark.range(1).select(F.current_user()).first()[0]
     _short = "".join(c if c.isalnum() else "_" for c in _user.split("@")[0])
     ml_schema = f"dev_{_short}_fraud"
+if not uc_model_name:
+    uc_model_name = f"{catalog_name}.{ml_schema}.fraud_detection"
+if not model_version:
+    import mlflow
+    from mlflow.tracking import MlflowClient
+
+    mlflow.set_registry_uri("databricks-uc")
+    _versions = [
+        int(mv.version)
+        for mv in MlflowClient().search_model_versions(f"name='{uc_model_name}'")
+    ]
+    if not _versions:
+        raise ValueError(
+            f"No model versions found for {uc_model_name!r}. Provide model_version explicitly."
+        )
+    model_version = str(max(_versions))
 
 endpoint_name = f"{ml_schema}_fraud"
 online_store_name = "fraud-workshop-online"
 card_feature_table = f"{catalog_name}.{ml_schema}.card_features"
 client_feature_table = f"{catalog_name}.{ml_schema}.client_features"
 
-print(f"Deploying {model_name} v{model_version} -> endpoint {endpoint_name}")
+print(f"Deploying {uc_model_name} v{model_version} -> endpoint {endpoint_name}")
 
 # COMMAND ----------
 
@@ -63,9 +87,9 @@
 mlflow.set_registry_uri("databricks-uc")
 client = MlflowClient()
 
-client.set_registered_model_alias(model_name, "champion", model_version)
+client.set_registered_model_alias(uc_model_name, "champion", model_version)
 try:
-    client.delete_registered_model_alias(model_name, "challenger")
+    client.delete_registered_model_alias(uc_model_name, "challenger")
 except Exception:
     pass  # there may be no challenger alias (e.g. first deployment)
 print(f"@champion -> version {model_version}")
@@ -131,7 +155,7 @@ def wait_online_store(name, minutes=20):
 
 served_entities = [
     ServedEntityInput(
-        entity_name=model_name,
+        entity_name=uc_model_name,
         entity_version=model_version,
         scale_to_zero_enabled=True,
         workload_size="Small",
diff --git a/src/deployment/model_deployment/notebooks/evaluation.py b/src/deployment/model_deployment/notebooks/evaluation.py
index 9d2003c..493150f 100644
--- a/src/deployment/model_deployment/notebooks/evaluation.py
+++ b/src/deployment/model_deployment/notebooks/evaluation.py
@@ -21,23 +21,58 @@
 
 # COMMAND ----------
 
-# The deployment job injects model_name + model_version as JOB-level parameters.
+# The deployment job injects the FULL three-level model name (catalog.schema.model) into
+# `model_name`, plus the `model_version` that triggered it. Both are empty when this notebook
+# is run interactively, so the fallback below rebuilds them.
 dbutils.widgets.text("model_name", "")
 dbutils.widgets.text("model_version", "")
 dbutils.widgets.text("catalog_name", "adoption_workshop")
+dbutils.widgets.text("ml_schema", "")
 dbutils.widgets.text("metric", "roc_auc")
 dbutils.widgets.text("baseline", "0.65")
 
-model_name = dbutils.widgets.get("model_name")
-model_version = dbutils.widgets.get("model_version")
+uc_model_name = dbutils.widgets.get("model_name").strip()
+model_version = dbutils.widgets.get("model_version").strip()
 catalog_name = dbutils.widgets.get("catalog_name")
+ml_schema = dbutils.widgets.get("ml_schema")
 metric = dbutils.widgets.get("metric")
 baseline = float(dbutils.widgets.get("baseline"))
 
-assert model_name and model_version, (
-    "model_name and model_version are injected by the deployment job."
-)
-print(f"Evaluating {model_name} version {model_version} on '{metric}' (floor {baseline}).")
+# Fail fast on a half-configured deployment job: model_name and model_version are injected
+# together. Both present = job; both empty = interactive run (derived below). Exactly one
+# present is a misconfiguration that would silently gate the wrong model or version.
+if bool(uc_model_name) != bool(model_version):
+    raise ValueError(
+        "Only one of model_name / model_version was provided. Pass both (deployment job) "
+        "or neither (interactive run)."
+    )
+
+from pyspark.sql import functions as F
+
+# Interactive fallback: rebuild the personal schema and the full model name the same way the
+# training/batch notebooks do, and default to the latest version, so this runs by hand too.
+if not ml_schema:
+    _user = spark.range(1).select(F.current_user()).first()[0]
+    _short = "".join(c if c.isalnum() else "_" for c in _user.split("@")[0])
+    ml_schema = f"dev_{_short}_fraud"
+if not uc_model_name:
+    uc_model_name = f"{catalog_name}.{ml_schema}.fraud_detection"
+if not model_version:
+    import mlflow
+    from mlflow.tracking import MlflowClient
+
+    mlflow.set_registry_uri("databricks-uc")
+    _versions = [
+        int(mv.version)
+        for mv in MlflowClient().search_model_versions(f"name='{uc_model_name}'")
+    ]
+    if not _versions:
+        raise ValueError(
+            f"No model versions found for {uc_model_name!r}. Provide model_version explicitly."
+        )
+    model_version = str(max(_versions))
+
+print(f"Evaluating {uc_model_name} version {model_version} on '{metric}' (floor {baseline}).")
 
 # COMMAND ----------
 
@@ -79,7 +114,7 @@
 
 def score_holdout(version):
     """Score a model version on the holdout; return a pandas DF of is_fraud + fraud score."""
-    scored = fe.score_batch(model_uri=f"models:/{model_name}/{version}", df=eval_spine)
+    scored = fe.score_batch(model_uri=f"models:/{uc_model_name}/{version}", df=eval_spine)
     return scored.select("is_fraud", F.col("prediction").cast("double").alias("score")).toPandas()
 
 
@@ -89,7 +124,7 @@ def score_holdout(version):
 # Fair comparison: re-score the current champion on the SAME holdout. If there is no champion
 # yet (first model), gate against the metric floor instead.
 try:
-    champion = client.get_model_version_by_alias(model_name, "champion")
+    champion = client.get_model_version_by_alias(uc_model_name, "champion")
     champ_pdf = score_holdout(champion.version)
     bar = float(roc_auc_score(champ_pdf["is_fraud"], champ_pdf["score"]))
     bar_label = f"champion (v{champion.version})"
@@ -99,9 +134,9 @@ def score_holdout(version):
 print(f"candidate {metric}={candidate_score:.4f}  vs  {bar_label}={bar:.4f}")
 
 # Surface the decision inputs on the model version so the approver sees them in the UI.
-client.set_model_version_tag(model_name, model_version, "eval_metric", metric)
-client.set_model_version_tag(model_name, model_version, "eval_score", f"{candidate_score:.4f}")
-client.set_model_version_tag(model_name, model_version, "eval_bar", f"{bar:.4f}")
+client.set_model_version_tag(uc_model_name, model_version, "eval_metric", metric)
+client.set_model_version_tag(uc_model_name, model_version, "eval_score", f"{candidate_score:.4f}")
+client.set_model_version_tag(uc_model_name, model_version, "eval_bar", f"{bar:.4f}")
 
 # COMMAND ----------
 
@@ -125,7 +160,7 @@ def score_holdout(version):
 ConfusionMatrixDisplay(cm, display_labels=["legit", "fraud"]).plot(
     cmap="Blues", ax=ax, colorbar=False
 )
-ax.set_title(f"{model_name.split('.')[-1]} v{model_version} (holdout, threshold 0.5)")
+ax.set_title(f"{uc_model_name.split('.')[-1]} v{model_version} (holdout, threshold 0.5)")
 fig.tight_layout()
 with mlflow.start_run(run_name=f"evaluation_v{model_version}") as ev_run:
     mlflow.log_figure(fig, "confusion_matrix.png")
diff --git a/src/training/notebooks/training.py b/src/training/notebooks/training.py
index 838d773..ac55125 100644
--- a/src/training/notebooks/training.py
+++ b/src/training/notebooks/training.py
@@ -25,7 +25,7 @@
 dbutils.widgets.text("gold_schema", "fraud_gold")
 dbutils.widgets.text("ml_schema", "")
 dbutils.widgets.text("experiment_name", "")
-dbutils.widgets.text("model_name", "")
+dbutils.widgets.text("model_name", "fraud_detection")
 dbutils.widgets.text("deployment_job_id", "")
 
 catalog_name = dbutils.widgets.get("catalog_name")
@@ -52,13 +52,17 @@
 experiment_name = (
     dbutils.widgets.get("experiment_name") or f"/Users/{current_user}/mlops-workshop-fraud"
 )
-model_name = dbutils.widgets.get("model_name") or f"{catalog_name}.{ml_schema}.fraud_detection"
+model_name = dbutils.widgets.get("model_name")
+# Accept either a bare name (prepend catalog + ml_schema) or an already-qualified 3-level name.
+uc_model_name = (
+    model_name if model_name.count(".") == 2 else f"{catalog_name}.{ml_schema}.{model_name}"
+)
 
 print(f"Card features:   {card_feature_table}")
 print(f"Client features: {client_feature_table}")
 print(f"Labels:          {source_table}")
 print(f"Experiment: {experiment_name}")
-print(f"Model:      {model_name}")
+print(f"Model:      {uc_model_name}")
 
 # COMMAND ----------
 
@@ -110,11 +114,11 @@
 
     _client = MlflowClient()
     try:
-        _client.get_registered_model(model_name)
-        _client.update_registered_model(model_name, deployment_job_id=deployment_job_id)
+        _client.get_registered_model(uc_model_name)
+        _client.update_registered_model(uc_model_name, deployment_job_id=deployment_job_id)
     except RestException:
-        _client.create_registered_model(model_name, deployment_job_id=deployment_job_id)
-    print(f"Connected {model_name} to deployment job {deployment_job_id}.")
+        _client.create_registered_model(uc_model_name, deployment_job_id=deployment_job_id)
+    print(f"Connected {uc_model_name} to deployment job {deployment_job_id}.")
 else:
     print("No deployment_job_id passed; skipping deployment-job connection (interactive run).")
 
@@ -299,7 +303,7 @@ def predict(self, context, model_input):
     # HINT:   reproduces the same features) and registers it to UC in a single call.
     # HINT: fe.log_model(model=FraudProbabilityModel(model), artifact_path="model",
     # HINT:   flavor=mlflow.pyfunc, training_set=training_set,
-    # HINT:   registered_model_name=model_name, infer_input_example=True)
+    # HINT:   registered_model_name=uc_model_name, infer_input_example=True)
     # <-- Your code here
 
     # MLflow 3: surface the eval metrics on the LoggedModel (and thus the UC model-version
@@ -389,10 +393,14 @@ def predict(self, context, model_input):
 client = MlflowClient()
 # Unity Catalog's search_model_versions only supports a `name='...'` filter (no run_id
 # filtering), so fetch this model's versions and take the highest: the one registered above.
-versions = client.search_model_versions(f"name='{model_name}'")
-new_version = max(int(mv.version) for mv in versions)
-client.set_registered_model_alias(model_name, "challenger", new_version)
-print(f"Registered {model_name} version {new_version} and set alias @challenger.")
+_versions = [int(mv.version) for mv in client.search_model_versions(f"name='{uc_model_name}'")]
+if not _versions:
+    raise ValueError(
+        f"No model versions found for {uc_model_name!r}. Was the model registered above?"
+    )
+new_version = max(_versions)
+client.set_registered_model_alias(uc_model_name, "challenger", new_version)
+print(f"Registered {uc_model_name} version {new_version} and set alias @challenger.")
 
 # COMMAND ----------