diff --git a/docs/anomalies.md b/docs/anomalies.md
index ad041db..c15bbf4 100644
--- a/docs/anomalies.md
+++ b/docs/anomalies.md
@@ -24,6 +24,12 @@ qualytics anomalies list --datastore-id 1 --status Active
 # Filter by date range
 qualytics anomalies list --datastore-id 1 \
   --start-date 2026-01-01 --end-date 2026-01-31
+
+# Only anomalies whose source records were enriched
+qualytics anomalies list --datastore-id 1 --source-enriched
+
+# Only anomalies whose source records were NOT enriched
+qualytics anomalies list --datastore-id 1 --no-source-enriched
 ```
 
 ## Inspecting an Anomaly
@@ -40,8 +46,19 @@ qualytics anomalies update --id 42 --status Acknowledged
 
 # Bulk update
 qualytics anomalies update --ids 42,43,44 --status Active
+
+# Set assignees on a single anomaly
+qualytics anomalies update --id 42 --status Active --assignee-ids "7,12"
+
+# Clear assignees (empty string)
+qualytics anomalies update --id 42 --status Active --assignee-ids ""
+
+# Bulk-assign a reviewer
+qualytics anomalies update --ids 42,43,44 --status Active --assignee-ids "7"
 ```
 
+User IDs come from `qualytics users list`.
+
 ## Archiving
 
 Archive anomalies with a resolution status:
diff --git a/docs/checks.md b/docs/checks.md
index 8a8aa43..bc074e5 100644
--- a/docs/checks.md
+++ b/docs/checks.md
@@ -81,6 +81,29 @@ qualytics checks create --datastore-id 1 --file checks.yaml
 | `tags` | list[string] | No | Tags for filtering |
 | `status` | string | No | `Active` or `Draft` (default: Active) |
 
+## Ownership and Assignment
+
+Each check can have an owner and a default anomaly assignee. Set them via the
+CLI flags `--owner-id` and `--default-anomaly-assignee-id` on `checks create`
+or `checks update`. Pass `0` on `update` to clear an existing value.
+
+User IDs are environment-specific, so ownership is **not** carried in the
+portable export YAML — `checks export` strips these fields and `checks
+import` won't restore them. Apply ownership at import time per environment.
+
+```bash
+# Apply a single owner to every check in a bulk import
+qualytics checks create --datastore-id 1 --file checks.yaml --owner-id 7
+
+# Update a check's assignee
+qualytics checks update --id 42 --file check.yaml --default-anomaly-assignee-id 12
+
+# Clear ownership on update
+qualytics checks update --id 42 --file check.yaml --owner-id 0
+```
+
+User IDs come from `qualytics users list`.
+
 ## Export and Import
 
 ### Export checks
diff --git a/docs/connections.md b/docs/connections.md
index 10f84a5..112890e 100644
--- a/docs/connections.md
+++ b/docs/connections.md
@@ -45,6 +45,32 @@ qualytics connections create \
   --parameters '{"role": "ANALYST", "warehouse": "COMPUTE_WH"}'
 ```
 
+### IAM Role authentication (S3, Athena, Redshift)
+
+S3, Athena, and Redshift connections can authenticate via an IAM Role instead
+of static credentials. Use `--authentication-type IAM_ROLE` together with
+`--role-arn` (and optionally `--external-id`).
+
+```bash
+# S3 with IAM Role (alternative to --access-key / --secret-key)
+qualytics connections create \
+  --type s3 \
+  --name s3-prod \
+  --uri s3://my-bucket \
+  --authentication-type IAM_ROLE \
+  --role-arn arn:aws:iam::123456789012:role/QualyticsReader \
+  --external-id my-external-id
+
+# Athena with IAM Role (alternative to --username / --password)
+qualytics connections create \
+  --type athena \
+  --name athena-prod \
+  --authentication-type IAM_ROLE \
+  --role-arn arn:aws:iam::123456789012:role/QualyticsAthena
+```
+
+Default authentication is `SHARED_KEY` for S3 and `BASIC` for Athena/Redshift.
+
 ## Listing and Retrieving
 
 ```bash
diff --git a/docs/operations.md b/docs/operations.md
index 5bf30eb..107ab38 100644
--- a/docs/operations.md
+++ b/docs/operations.md
@@ -1,13 +1,13 @@
 # Operations
 
-Operations are the data processing workflows in Qualytics. The standard lifecycle is: **sync** (discover containers) then **profile** (infer checks) then **scan** (detect anomalies).
+Operations are the data processing workflows in Qualytics. The standard lifecycle is: **sync** (discover containers) then **profile** (generate AI managed checks) then **scan** (detect anomalies).
 
 ## Commands
 
 | Command | Description |
 |---------|-------------|
 | `operations sync` | Trigger a sync operation (discover containers) |
-| `operations profile` | Trigger a profile operation (infer quality checks) |
+| `operations profile` | Trigger a profile operation (generate AI managed checks) |
 | `operations scan` | Trigger a scan operation (detect anomalies) |
 | `operations materialize` | Trigger a materialize operation (computed containers) |
 | `operations export` | Trigger an export operation (anomalies, checks, profiles) |
@@ -25,16 +25,16 @@ qualytics operations sync --datastore-id 1
 
 Discovers tables, views, and files in the datastore and creates container records.
 
-### 2. Profile (infer checks)
+### 2. Profile (generate AI managed checks)
 
 ```bash
 qualytics operations profile --datastore-id 1
 
-# With inference threshold (higher = more checks inferred)
-qualytics operations profile --datastore-id 1 --inference-threshold 3
+# AI Effort levels: off, low, medium, high, xhigh, max
+qualytics operations profile --datastore-id 1 --ai-effort high
 ```
 
-Profiles container data to infer quality checks based on statistical analysis.
+Profiles container data and generates AI managed checks based on statistical analysis.
 
 ### 3. Scan (detect anomalies)
 
@@ -46,9 +46,15 @@ qualytics operations scan --datastore-id 1 --container-names "orders,customers"
 
 # Incremental scan (only new/updated records)
 qualytics operations scan --datastore-id 1 --incremental
+
+# Disable auto-resolution of passed anomalies (server default is on)
+qualytics operations scan --datastore-id 1 --no-auto-resolve-passed-anomalies
 ```
 
-Runs quality checks against the data and detects anomalies.
+Runs quality checks against the data and detects anomalies. When a scan
+completes, open anomalies whose fingerprints no longer fail are auto-resolved
+unless `--no-auto-resolve-passed-anomalies` is passed. Auto-resolution is
+silently disabled for incremental scans regardless of the flag.
 
 ## Running in Background
 
diff --git a/qualytics/api/anomalies.py b/qualytics/api/anomalies.py
index f7a07e5..4864eeb 100644
--- a/qualytics/api/anomalies.py
+++ b/qualytics/api/anomalies.py
@@ -19,6 +19,7 @@ def list_anomalies(
     archived: str | None = None,
     sort_created: str | None = None,
     sort_weight: str | None = None,
+    source_enriched: bool | None = None,
     page: int = 1,
     size: int = 100,
 ) -> dict:
@@ -53,6 +54,8 @@ def list_anomalies(
         params["sort_created"] = sort_created
     if sort_weight:
         params["sort_weight"] = sort_weight
+    if source_enriched is not None:
+        params["source_enriched"] = str(source_enriched).lower()
     response = client.get("anomalies", params=params)
     return response.json()
 
diff --git a/qualytics/api/client.py b/qualytics/api/client.py
index 91779c8..91f78e3 100644
--- a/qualytics/api/client.py
+++ b/qualytics/api/client.py
@@ -1,9 +1,33 @@
 """Centralized API client for the Qualytics controlplane."""
 
+import os
+
 import requests
 import urllib3
 from rich import print
 
+DEFAULT_TIMEOUT = 30
+
+
+def _resolve_timeout(config: dict | None) -> int:
+    """Resolve request timeout from env var, config, or default.
+
+    Precedence: ``QUALYTICS_TIMEOUT`` env var → ``timeout`` in config → 30.
+    """
+    env_value = os.environ.get("QUALYTICS_TIMEOUT")
+    if env_value:
+        try:
+            parsed = int(env_value)
+            if parsed > 0:
+                return parsed
+        except ValueError:
+            pass
+    if config:
+        cfg_value = config.get("timeout")
+        if isinstance(cfg_value, int) and cfg_value > 0:
+            return cfg_value
+    return DEFAULT_TIMEOUT
+
 
 class QualyticsAPIError(Exception):
     """Base exception for API errors."""
@@ -51,7 +75,7 @@ def __init__(
         base_url: str,
         token: str,
         ssl_verify: bool = True,
-        timeout: int = 30,
+        timeout: int = DEFAULT_TIMEOUT,
     ):
         self.base_url = base_url.rstrip("/")
         if not self.base_url.endswith("/"):
@@ -109,6 +133,12 @@ def _request(self, method: str, path: str, **kwargs) -> requests.Response:
                 "If your server runs plain HTTP, use http:// instead of https:// "
                 "in your URL (e.g. qualytics init --url http://localhost:8000)."
             )
+        except requests.exceptions.Timeout:
+            raise ConnectionError(
+                f"Request to {url} timed out after {kwargs['timeout']}s. "
+                "Increase the timeout via QUALYTICS_TIMEOUT=<seconds> "
+                "or set 'timeout' in ~/.qualytics/config.yaml."
+            )
         except requests.exceptions.ConnectionError:
             raise ConnectionError(
                 f"Could not connect to {url}. "
@@ -180,4 +210,5 @@ def get_client(config: dict | None = None) -> QualyticsClient:
         base_url=base_url,
         token=token,
         ssl_verify=ssl_verify,
+        timeout=_resolve_timeout(config),
     )
diff --git a/qualytics/cli/anomalies.py b/qualytics/cli/anomalies.py
index d43f791..604d01d 100644
--- a/qualytics/cli/anomalies.py
+++ b/qualytics/cli/anomalies.py
@@ -78,6 +78,11 @@ def anomalies_list(
     end_date: str | None = typer.Option(
         None, "--end-date", help="End date (YYYY-MM-DD)"
     ),
+    source_enriched: bool | None = typer.Option(
+        None,
+        "--source-enriched/--no-source-enriched",
+        help="Filter by source-record enrichment status (omit flag for no filter)",
+    ),
     fmt: OutputFormat = typer.Option(
         OutputFormat.YAML, "--format", help="Output format: yaml or json"
     ),
@@ -116,6 +121,7 @@ def anomalies_list(
             start_date=start_date,
             end_date=end_date,
             archived=archived,
+            source_enriched=source_enriched,
         )
 
     print(f"[green]Found {len(all_anomalies)} anomalies.[/green]")
@@ -142,6 +148,11 @@ def anomalies_update(
         None, "--description", help="Update description"
     ),
     tags: str | None = typer.Option(None, "--tags", help="Comma-separated tag names"),
+    assignee_ids: str | None = typer.Option(
+        None,
+        "--assignee-ids",
+        help='Comma-separated assignee user IDs (e.g. "1,2"). Empty string clears assignees.',
+    ),
 ):
     """Update anomaly status (Active or Acknowledged)."""
     if not anomaly_id and not ids:
@@ -156,6 +167,12 @@ def anomalies_update(
         )
         raise typer.Exit(code=1)
 
+    parsed_assignees: list[int] | None = None
+    if assignee_ids is not None:
+        parsed_assignees = (
+            [int(x) for x in _parse_comma_list(assignee_ids)] if assignee_ids else []
+        )
+
     client = get_client()
 
     if anomaly_id and not ids:
@@ -165,6 +182,8 @@ def anomalies_update(
             payload["description"] = description
         if tags:
             payload["tags"] = _parse_comma_list(tags)
+        if parsed_assignees is not None:
+            payload["assignee_ids"] = parsed_assignees
         result = update_anomaly(client, anomaly_id, payload)
         print(f"[green]Anomaly {result['id']} updated to '{status}'.[/green]")
     else:
@@ -175,7 +194,10 @@ def anomalies_update(
         if ids:
             id_list.extend(int(x) for x in _parse_comma_list(ids))
 
-        items = [{"id": aid, "status": status} for aid in id_list]
+        item_template: dict = {"status": status}
+        if parsed_assignees is not None:
+            item_template["assignee_ids"] = parsed_assignees
+        items = [{"id": aid, **item_template} for aid in id_list]
         bulk_update_anomalies(client, items)
         print(f"[green]Updated {len(id_list)} anomalies to '{status}'.[/green]")
 
diff --git a/qualytics/cli/auth.py b/qualytics/cli/auth.py
index 5b95943..70317e9 100644
--- a/qualytics/cli/auth.py
+++ b/qualytics/cli/auth.py
@@ -183,6 +183,10 @@ def auth_status():
     )
     ssl_label = "[green]enabled[/green]" if ssl_verify else "[yellow]disabled[/yellow]"
 
+    from ..api.client import _resolve_timeout
+
+    timeout = _resolve_timeout(config)
+
     print(f"[bold]{host}[/bold]")
     print(f"  URL:              {url}")
     print(f"  Status:           {status_icon}")
@@ -190,6 +194,7 @@ def auth_status():
     if expiry_line:
         print(f"  Expiry:           {expiry_line}")
     print(f"  SSL Verification: {ssl_label}")
+    print(f"  Request timeout:  {timeout}s")
     print(f"  Config file:      {CONFIG_PATH}")
 
     if not token_valid:
diff --git a/qualytics/cli/checks.py b/qualytics/cli/checks.py
index db00803..da030ae 100644
--- a/qualytics/cli/checks.py
+++ b/qualytics/cli/checks.py
@@ -59,6 +59,16 @@ def checks_create(
     file: str = typer.Option(
         ..., "--file", "-f", help="YAML/JSON file with check definition(s)"
     ),
+    owner_id: int | None = typer.Option(
+        None,
+        "--owner-id",
+        help="Apply this owner user ID to every check in the batch (overrides file)",
+    ),
+    default_anomaly_assignee_id: int | None = typer.Option(
+        None,
+        "--default-anomaly-assignee-id",
+        help="Apply this default anomaly assignee user ID to every check (overrides file)",
+    ),
 ):
     """Create quality checks from a file (single or bulk)."""
     client = get_client()
@@ -102,6 +112,10 @@ def checks_create(
                 failed += 1
                 continue
         try:
+            if owner_id is not None:
+                check["owner_id"] = owner_id
+            if default_anomaly_assignee_id is not None:
+                check["default_anomaly_assignee_id"] = default_anomaly_assignee_id
             payload = _build_create_payload(check, container_id)
             result = create_quality_check(client, payload)
             print(
@@ -198,6 +212,16 @@ def checks_update(
     file: str = typer.Option(
         ..., "--file", "-f", help="YAML/JSON file with updated check definition"
     ),
+    owner_id: int | None = typer.Option(
+        None,
+        "--owner-id",
+        help="Owner user ID (overrides file). Pass 0 to clear.",
+    ),
+    default_anomaly_assignee_id: int | None = typer.Option(
+        None,
+        "--default-anomaly-assignee-id",
+        help="Default anomaly assignee user ID (overrides file). Pass 0 to clear.",
+    ),
 ):
     """Update a quality check from a file."""
     client = get_client()
@@ -214,6 +238,21 @@ def checks_update(
         "status": data.get("status", "Active"),
     }
 
+    # CLI flag (if given) takes precedence over the file value.
+    # Pass through to the same normalization the service uses (0 → clear).
+    effective_owner = owner_id if owner_id is not None else data.get("owner_id")
+    effective_assignee = (
+        default_anomaly_assignee_id
+        if default_anomaly_assignee_id is not None
+        else data.get("default_anomaly_assignee_id")
+    )
+    if effective_owner is not None:
+        payload["owner_id"] = effective_owner if effective_owner else None
+    if effective_assignee is not None:
+        payload["default_anomaly_assignee_id"] = (
+            effective_assignee if effective_assignee else None
+        )
+
     result = update_quality_check(client, check_id, payload)
     print(f"[green]Quality check {result['id']} updated successfully.[/green]")
 
diff --git a/qualytics/cli/connections.py b/qualytics/cli/connections.py
index 03678dd..342c93c 100644
--- a/qualytics/cli/connections.py
+++ b/qualytics/cli/connections.py
@@ -123,6 +123,25 @@ def connections_create(
     max_parallelization: int | None = typer.Option(
         None, "--max-parallelization", help="Max parallelization level"
     ),
+    authentication_type: str | None = typer.Option(
+        None,
+        "--authentication-type",
+        help=(
+            "Authentication mode for S3/Athena/Redshift. "
+            "S3: SHARED_KEY (default) or IAM_ROLE. "
+            "Athena/Redshift: BASIC (default) or IAM_ROLE."
+        ),
+    ),
+    role_arn: str | None = typer.Option(
+        None,
+        "--role-arn",
+        help="IAM Role ARN (required when --authentication-type IAM_ROLE)",
+    ),
+    external_id: str | None = typer.Option(
+        None,
+        "--external-id",
+        help="Optional external ID for IAM Role assumption",
+    ),
     parameters: str | None = typer.Option(
         None,
         "--parameters",
@@ -168,21 +187,28 @@ def connections_create(
             print(f"[red]Invalid JSON in --parameters: {e}[/red]")
             raise typer.Exit(code=1)
 
-    payload = build_create_connection_payload(
-        connection_type,
-        name=name,
-        host=resolved.get("host", host),
-        port=port,
-        username=resolved.get("username", username),
-        password=resolved.get("password"),
-        uri=resolved.get("uri"),
-        access_key=resolved.get("access_key"),
-        secret_key=resolved.get("secret_key"),
-        catalog=catalog,
-        jdbc_fetch_size=jdbc_fetch_size,
-        max_parallelization=max_parallelization,
-        parameters=extra_params,
-    )
+    try:
+        payload = build_create_connection_payload(
+            connection_type,
+            name=name,
+            host=resolved.get("host", host),
+            port=port,
+            username=resolved.get("username", username),
+            password=resolved.get("password"),
+            uri=resolved.get("uri"),
+            access_key=resolved.get("access_key"),
+            secret_key=resolved.get("secret_key"),
+            catalog=catalog,
+            jdbc_fetch_size=jdbc_fetch_size,
+            max_parallelization=max_parallelization,
+            authentication_type=authentication_type,
+            role_arn=role_arn,
+            external_id=external_id,
+            parameters=extra_params,
+        )
+    except ValueError as e:
+        print(f"[red]{e}[/red]")
+        raise typer.Exit(code=1)
 
     print("[bold]Connection Create Payload (secrets redacted):[/bold]")
     print(format_for_display(redact_payload(payload), fmt))
@@ -223,6 +249,15 @@ def connections_update(
     secret_key: str | None = typer.Option(
         None, "--secret-key", help="New secret key (supports ${ENV_VAR})"
     ),
+    authentication_type: str | None = typer.Option(
+        None,
+        "--authentication-type",
+        help="New authentication mode (S3/Athena/Redshift): SHARED_KEY, BASIC, or IAM_ROLE",
+    ),
+    role_arn: str | None = typer.Option(None, "--role-arn", help="New IAM Role ARN"),
+    external_id: str | None = typer.Option(
+        None, "--external-id", help="New IAM Role external ID"
+    ),
     parameters: str | None = typer.Option(
         None,
         "--parameters",
@@ -255,16 +290,23 @@ def connections_update(
             print(f"[red]Invalid JSON in --parameters: {e}[/red]")
             raise typer.Exit(code=1)
 
-    changes = build_update_connection_payload(
-        name=name,
-        host=resolved.get("host"),
-        port=port,
-        username=resolved.get("username"),
-        password=resolved.get("password"),
-        uri=resolved.get("uri"),
-        access_key=resolved.get("access_key"),
-        secret_key=resolved.get("secret_key"),
-    )
+    try:
+        changes = build_update_connection_payload(
+            name=name,
+            host=resolved.get("host"),
+            port=port,
+            username=resolved.get("username"),
+            password=resolved.get("password"),
+            uri=resolved.get("uri"),
+            access_key=resolved.get("access_key"),
+            secret_key=resolved.get("secret_key"),
+            authentication_type=authentication_type,
+            role_arn=role_arn,
+            external_id=external_id,
+        )
+    except ValueError as e:
+        print(f"[red]{e}[/red]")
+        raise typer.Exit(code=1)
 
     # Merge extra parameters
     if extra_params:
diff --git a/qualytics/cli/operations.py b/qualytics/cli/operations.py
index 09b47cb..ee2946a 100644
--- a/qualytics/cli/operations.py
+++ b/qualytics/cli/operations.py
@@ -29,6 +29,20 @@
 
 _VALID_REMEDIATION = {"append", "overwrite", "none"}
 _VALID_ASSET_TYPES = {"anomalies", "checks", "profiles"}
+_VALID_AI_EFFORT = {
+    "off",
+    "low",
+    "medium",
+    "high",
+    "xhigh",
+    "max",
+    "0",
+    "1",
+    "2",
+    "3",
+    "4",
+    "5",
+}
 
 
 # ── helpers ──────────────────────────────────────────────────────────────
@@ -121,15 +135,21 @@ def profile_operation(
         "--container-tags",
         help='Comma-separated container tags. Example: "production,finance"',
     ),
-    inference_threshold: int | None = typer.Option(
+    ai_effort: str | None = typer.Option(
+        None,
+        "--ai-effort",
+        help="AI Effort level: off, low, medium, high, xhigh, max",
+    ),
+    inference_threshold: str | None = typer.Option(
         None,
         "--inference-threshold",
-        help="Inference quality checks threshold (0 to 5)",
+        help="Deprecated: use --ai-effort",
+        hidden=True,
     ),
     infer_as_draft: bool = typer.Option(
         False,
         "--infer-as-draft",
-        help="Infer all quality checks as Draft",
+        help="Create AI managed checks as draft",
     ),
     max_records_analyzed_per_partition: int | None = typer.Option(
         None,
@@ -139,7 +159,7 @@ def profile_operation(
     max_count_testing_sample: int | None = typer.Option(
         None,
         "--max-count-testing-sample",
-        help="Records accumulated for validation of inferred checks (max 100000)",
+        help="Records accumulated for validation of AI managed checks (max 100000)",
     ),
     percent_testing_threshold: float | None = typer.Option(
         None, "--percent-testing-threshold", help="Percent of testing threshold"
@@ -182,6 +202,21 @@ def profile_operation(
     datastore_ids = _parse_int_list(datastore_id)
     client = get_client()
 
+    if inference_threshold is not None:
+        print(
+            "[bold yellow]--inference-threshold is deprecated, use --ai-effort instead.[/bold yellow]"
+        )
+        if ai_effort is None:
+            ai_effort = inference_threshold
+
+    if ai_effort is not None:
+        ai_effort = ai_effort.lower()
+        if ai_effort not in _VALID_AI_EFFORT:
+            print(
+                "[bold red]--ai-effort must be one of: off, low, medium, high, xhigh, max[/bold red]"
+            )
+            raise typer.Exit(code=1)
+
     if (
         max_records_analyzed_per_partition is not None
         and max_records_analyzed_per_partition < -1
@@ -204,7 +239,7 @@ def profile_operation(
         datastore_ids=datastore_ids,
         container_names=names_list,
         container_tags=tags_list,
-        inference_threshold=inference_threshold,
+        ai_effort=ai_effort,
         infer_as_draft=infer_as_draft if infer_as_draft else None,
         max_records_analyzed_per_partition=max_records_analyzed_per_partition,
         max_count_testing_sample=max_count_testing_sample,
@@ -259,6 +294,14 @@ def scan_operation(
         "--enrichment-source-record-limit",
         help="Limit of enrichment source records per run (>= 1)",
     ),
+    auto_resolve_passed_anomalies: bool | None = typer.Option(
+        None,
+        "--auto-resolve-passed-anomalies/--no-auto-resolve-passed-anomalies",
+        help=(
+            "Auto-resolve open anomalies whose fingerprint no longer fails. "
+            "Server default is on. Silently forced off for incremental scans."
+        ),
+    ),
     greater_than_time: datetime | None = typer.Option(
         None,
         "--greater-than-time",
@@ -326,6 +369,7 @@ def scan_operation(
         remediation=remediation,
         max_records_analyzed_per_partition=max_records_analyzed_per_partition,
         enrichment_source_record_limit=enrichment_source_record_limit,
+        auto_resolve_passed_anomalies=auto_resolve_passed_anomalies,
         greater_than_time=gt_time,
         greater_than_batch=greater_than_batch,
         background=background,
diff --git a/qualytics/services/connections.py b/qualytics/services/connections.py
index cc4ed58..3e9278f 100644
--- a/qualytics/services/connections.py
+++ b/qualytics/services/connections.py
@@ -74,6 +74,9 @@ def build_create_connection_payload(
     catalog: str | None = None,
     jdbc_fetch_size: int | None = None,
     max_parallelization: int | None = None,
+    authentication_type: str | None = None,
+    role_arn: str | None = None,
+    external_id: str | None = None,
     parameters: dict | None = None,
 ) -> dict:
     """Build a payload for creating a connection.
@@ -115,7 +118,17 @@ def build_create_connection_payload(
     if max_parallelization is not None:
         payload["max_parallelization"] = max_parallelization
 
-    # Merge the catch-all parameters dict last (overrides dedicated flags)
+    # IAM Role auth (S3, Athena, Redshift) — these go *inside* the
+    # ``parameters`` dict on the wire (controlplane spec uses
+    # ``map_to="parameters"`` for them), not at the top level.
+    _require_role_arn_for_iam_role(authentication_type, role_arn)
+    iam_params = _iam_role_params(authentication_type, role_arn, external_id)
+    if iam_params:
+        payload["parameters"] = {**(payload.get("parameters") or {}), **iam_params}
+
+    # Merge the catch-all parameters dict last (overrides dedicated flags).
+    # Top-level merge is preserved for legacy callers that used --parameters
+    # to set fields like Snowflake's role/warehouse.
     if parameters is not None:
         payload.update(parameters)
 
@@ -125,12 +138,53 @@ def build_create_connection_payload(
 def build_update_connection_payload(**changes) -> dict:
     """Build a partial-update payload for a connection.
 
-    Only non-None values are included.
+    Only non-None values are included. IAM Role fields (``authentication_type``,
+    ``role_arn``, ``external_id``) are nested under ``parameters``.
     """
-    payload: dict = {}
+    iam_keys = {"authentication_type", "role_arn", "external_id"}
+    iam_changes = {k: changes.pop(k) for k in list(changes) if k in iam_keys}
+
+    _require_role_arn_for_iam_role(
+        iam_changes.get("authentication_type"), iam_changes.get("role_arn")
+    )
 
+    payload: dict = {}
     for key, value in changes.items():
         if value is not None:
             payload[key] = value
 
+    iam_params = _iam_role_params(
+        iam_changes.get("authentication_type"),
+        iam_changes.get("role_arn"),
+        iam_changes.get("external_id"),
+    )
+    if iam_params:
+        payload["parameters"] = iam_params
+
     return payload
+
+
+def _iam_role_params(
+    authentication_type: str | None,
+    role_arn: str | None,
+    external_id: str | None,
+) -> dict:
+    """Collect non-None IAM Role fields into a ``parameters`` sub-dict."""
+    out: dict = {}
+    if authentication_type is not None:
+        out["authentication_type"] = authentication_type
+    if role_arn is not None:
+        out["role_arn"] = role_arn
+    if external_id is not None:
+        out["external_id"] = external_id
+    return out
+
+
+def _require_role_arn_for_iam_role(
+    authentication_type: str | None, role_arn: str | None
+) -> None:
+    """Fail fast if IAM_ROLE is selected without a role ARN."""
+    if authentication_type == "IAM_ROLE" and not role_arn:
+        raise ValueError(
+            "--role-arn is required when --authentication-type is IAM_ROLE."
+        )
diff --git a/qualytics/services/operations.py b/qualytics/services/operations.py
index 6dfc147..c85d018 100644
--- a/qualytics/services/operations.py
+++ b/qualytics/services/operations.py
@@ -17,6 +17,23 @@
 from ..config import OPERATION_ERROR_PATH
 from ..utils.file_ops import log_error
 
+_INT_TO_AI_EFFORT = {
+    "0": "off",
+    "1": "low",
+    "2": "medium",
+    "3": "high",
+    "4": "xhigh",
+    "5": "max",
+}
+
+
+def _normalize_ai_effort(value: str | None) -> str | None:
+    """Convert legacy numeric string (e.g. '3') to the label form ('high'). Labels pass through unchanged."""
+    if value is None:
+        return None
+    return _INT_TO_AI_EFFORT.get(value, value)
+
+
 # Default polling configuration
 DEFAULT_POLL_INTERVAL = 10  # seconds between polls
 DEFAULT_TIMEOUT = 1800  # 30 minutes max wait
@@ -213,7 +230,7 @@ def run_profile(
     datastore_ids: list[int],
     container_names: list[str] | None,
     container_tags: list[str] | None,
-    inference_threshold: int | None,
+    ai_effort: str | None,
     infer_as_draft: bool | None,
     max_records_analyzed_per_partition: int | None,
     max_count_testing_sample: int | None,
@@ -227,6 +244,7 @@ def run_profile(
     timeout: int = DEFAULT_TIMEOUT,
 ):
     """Run profile operation for specified datastores."""
+    ai_effort = _normalize_ai_effort(ai_effort)
 
     def build_payload(datastore_id):
         return {
@@ -234,7 +252,7 @@ def build_payload(datastore_id):
             "type": "profile",
             "container_names": container_names,
             "container_tags": container_tags,
-            "inference_threshold": inference_threshold,
+            "ai_effort": ai_effort,
             "infer_as_draft": infer_as_draft,
             "max_records_analyzed_per_partition": max_records_analyzed_per_partition,
             "max_count_testing_sample": max_count_testing_sample,
@@ -268,6 +286,7 @@ def run_scan(
     greater_than_time: str | None,
     greater_than_batch: float | None,
     background: bool,
+    auto_resolve_passed_anomalies: bool | None = None,
     poll_interval: int = DEFAULT_POLL_INTERVAL,
     timeout: int = DEFAULT_TIMEOUT,
 ):
@@ -279,6 +298,7 @@ def build_payload(datastore_id):
             "type": "scan",
             "incremental": incremental if incremental is not None else False,
             "remediation": remediation,
+            "auto_resolve_passed_anomalies": auto_resolve_passed_anomalies,
             "max_records_analyzed_per_partition": max_records_analyzed_per_partition,
             "enrichment_source_record_limit": enrichment_source_record_limit,
             "greater_than_time": greater_than_time,
diff --git a/qualytics/services/quality_checks.py b/qualytics/services/quality_checks.py
index 3fa0863..f026d2f 100644
--- a/qualytics/services/quality_checks.py
+++ b/qualytics/services/quality_checks.py
@@ -188,9 +188,16 @@ def _build_uid_lookup(client: QualyticsClient, datastore_id: int) -> dict[str, i
     return lookup
 
 
+def _user_ref(value):
+    """Normalize a user-id reference: 0 (or any falsy non-None) means clear."""
+    if value is None:
+        return ...  # sentinel: caller should not include the key
+    return value if value else None
+
+
 def _build_create_payload(check: dict, container_id: int) -> dict:
     """Convert a portable check dict into a POST /quality-checks payload."""
-    return {
+    payload = {
         "container_id": container_id,
         "rule": check.get("rule_type") or check.get("rule", ""),
         "description": check.get("description", ""),
@@ -202,11 +209,18 @@ def _build_create_payload(check: dict, container_id: int) -> dict:
         "additional_metadata": check.get("additional_metadata") or {},
         "status": check.get("status", "Active"),
     }
+    owner = _user_ref(check.get("owner_id"))
+    if owner is not ...:
+        payload["owner_id"] = owner
+    assignee = _user_ref(check.get("default_anomaly_assignee_id"))
+    if assignee is not ...:
+        payload["default_anomaly_assignee_id"] = assignee
+    return payload
 
 
 def _build_update_payload(check: dict) -> dict:
     """Convert a portable check dict into a PUT /quality-checks/{id} payload."""
-    return {
+    payload = {
         "description": check.get("description", ""),
         "fields": check.get("fields") or [],
         "coverage": check.get("coverage"),
@@ -216,6 +230,13 @@ def _build_update_payload(check: dict) -> dict:
         "additional_metadata": check.get("additional_metadata") or {},
         "status": check.get("status", "Active"),
     }
+    owner = _user_ref(check.get("owner_id"))
+    if owner is not ...:
+        payload["owner_id"] = owner
+    assignee = _user_ref(check.get("default_anomaly_assignee_id"))
+    if assignee is not ...:
+        payload["default_anomaly_assignee_id"] = assignee
+    return payload
 
 
 def import_checks_to_datastore(
diff --git a/tests/test_anomalies.py b/tests/test_anomalies.py
index 9008579..0b9b0ce 100644
--- a/tests/test_anomalies.py
+++ b/tests/test_anomalies.py
@@ -75,6 +75,21 @@ def test_none_filters_excluded(self):
         assert "container" not in params
         assert "status" not in params
         assert "tag" not in params
+        assert "source_enriched" not in params
+
+    def test_source_enriched_true(self):
+        client = _mock_client()
+        client.get.return_value.json.return_value = {"items": [], "total": 0}
+        list_anomalies(client, source_enriched=True)
+        params = client.get.call_args.kwargs["params"]
+        assert params["source_enriched"] == "true"
+
+    def test_source_enriched_false(self):
+        client = _mock_client()
+        client.get.return_value.json.return_value = {"items": [], "total": 0}
+        list_anomalies(client, source_enriched=False)
+        params = client.get.call_args.kwargs["params"]
+        assert params["source_enriched"] == "false"
 
 
 class TestListAllAnomalies:
@@ -247,6 +262,44 @@ def test_list_with_filters(self, mock_gc, mock_list, cli_runner):
         assert kwargs["quality_check"] == 5
         assert kwargs["anomaly_type"] == "record"
 
+    @patch("qualytics.cli.anomalies.list_all_anomalies")
+    @patch("qualytics.cli.anomalies.get_client")
+    def test_list_with_source_enriched(self, mock_gc, mock_list, cli_runner):
+        mock_gc.return_value = _mock_client()
+        mock_list.return_value = []
+        result = cli_runner.invoke(
+            app,
+            ["anomalies", "list", "--datastore-id", "42", "--source-enriched"],
+        )
+        assert result.exit_code == 0
+        _, kwargs = mock_list.call_args
+        assert kwargs["source_enriched"] is True
+
+    @patch("qualytics.cli.anomalies.list_all_anomalies")
+    @patch("qualytics.cli.anomalies.get_client")
+    def test_list_with_no_source_enriched(self, mock_gc, mock_list, cli_runner):
+        mock_gc.return_value = _mock_client()
+        mock_list.return_value = []
+        result = cli_runner.invoke(
+            app,
+            ["anomalies", "list", "--datastore-id", "42", "--no-source-enriched"],
+        )
+        assert result.exit_code == 0
+        _, kwargs = mock_list.call_args
+        assert kwargs["source_enriched"] is False
+
+    @patch("qualytics.cli.anomalies.list_all_anomalies")
+    @patch("qualytics.cli.anomalies.get_client")
+    def test_list_default_no_source_enriched_filter(
+        self, mock_gc, mock_list, cli_runner
+    ):
+        mock_gc.return_value = _mock_client()
+        mock_list.return_value = []
+        result = cli_runner.invoke(app, ["anomalies", "list", "--datastore-id", "42"])
+        assert result.exit_code == 0
+        _, kwargs = mock_list.call_args
+        assert kwargs["source_enriched"] is None
+
     @patch("qualytics.cli.anomalies.list_all_anomalies")
     @patch("qualytics.cli.anomalies.get_client")
     def test_list_with_date_range(self, mock_gc, mock_list, cli_runner):
@@ -312,6 +365,73 @@ def test_update_no_id_or_ids(self, cli_runner):
             )
         assert result.exit_code == 1
 
+    @patch("qualytics.cli.anomalies.update_anomaly")
+    @patch("qualytics.cli.anomalies.get_client")
+    def test_update_with_assignee_ids(self, mock_gc, mock_update, cli_runner):
+        mock_gc.return_value = _mock_client()
+        mock_update.return_value = {"id": 42, "status": "Active"}
+        result = cli_runner.invoke(
+            app,
+            [
+                "anomalies",
+                "update",
+                "--id",
+                "42",
+                "--status",
+                "Active",
+                "--assignee-ids",
+                "7,12",
+            ],
+        )
+        assert result.exit_code == 0
+        payload = mock_update.call_args.args[2]
+        assert payload["assignee_ids"] == [7, 12]
+
+    @patch("qualytics.cli.anomalies.update_anomaly")
+    @patch("qualytics.cli.anomalies.get_client")
+    def test_update_clears_assignees_with_empty_string(
+        self, mock_gc, mock_update, cli_runner
+    ):
+        mock_gc.return_value = _mock_client()
+        mock_update.return_value = {"id": 42, "status": "Active"}
+        result = cli_runner.invoke(
+            app,
+            [
+                "anomalies",
+                "update",
+                "--id",
+                "42",
+                "--status",
+                "Active",
+                "--assignee-ids",
+                "",
+            ],
+        )
+        assert result.exit_code == 0
+        payload = mock_update.call_args.args[2]
+        assert payload["assignee_ids"] == []
+
+    @patch("qualytics.cli.anomalies.bulk_update_anomalies")
+    @patch("qualytics.cli.anomalies.get_client")
+    def test_update_bulk_with_assignee_ids(self, mock_gc, mock_bulk, cli_runner):
+        mock_gc.return_value = _mock_client()
+        result = cli_runner.invoke(
+            app,
+            [
+                "anomalies",
+                "update",
+                "--ids",
+                "1,2",
+                "--status",
+                "Active",
+                "--assignee-ids",
+                "7",
+            ],
+        )
+        assert result.exit_code == 0
+        items = mock_bulk.call_args.args[1]
+        assert all(item["assignee_ids"] == [7] for item in items)
+
     @patch("qualytics.cli.anomalies.update_anomaly")
     @patch("qualytics.cli.anomalies.get_client")
     def test_update_with_description_and_tags(self, mock_gc, mock_update, cli_runner):
diff --git a/tests/test_client.py b/tests/test_client.py
index b6cced7..6376b29 100644
--- a/tests/test_client.py
+++ b/tests/test_client.py
@@ -265,3 +265,39 @@ def test_get_client_respects_ssl_verify(self):
         }
         client = get_client(config)
         assert client.ssl_verify is False
+
+    def test_get_client_default_timeout(self, monkeypatch):
+        import jwt
+
+        monkeypatch.delenv("QUALYTICS_TIMEOUT", raising=False)
+        token = jwt.encode({"sub": "user"}, key="", algorithm="HS256")
+        config = {"url": "https://example.com/api", "token": token}
+        client = get_client(config)
+        assert client.timeout == 30
+
+    def test_get_client_timeout_from_config(self, monkeypatch):
+        import jwt
+
+        monkeypatch.delenv("QUALYTICS_TIMEOUT", raising=False)
+        token = jwt.encode({"sub": "user"}, key="", algorithm="HS256")
+        config = {"url": "https://example.com/api", "token": token, "timeout": 90}
+        client = get_client(config)
+        assert client.timeout == 90
+
+    def test_get_client_timeout_from_env_overrides_config(self, monkeypatch):
+        import jwt
+
+        monkeypatch.setenv("QUALYTICS_TIMEOUT", "120")
+        token = jwt.encode({"sub": "user"}, key="", algorithm="HS256")
+        config = {"url": "https://example.com/api", "token": token, "timeout": 90}
+        client = get_client(config)
+        assert client.timeout == 120
+
+    def test_get_client_timeout_env_invalid_falls_back(self, monkeypatch):
+        import jwt
+
+        monkeypatch.setenv("QUALYTICS_TIMEOUT", "not-a-number")
+        token = jwt.encode({"sub": "user"}, key="", algorithm="HS256")
+        config = {"url": "https://example.com/api", "token": token, "timeout": 60}
+        client = get_client(config)
+        assert client.timeout == 60
diff --git a/tests/test_connections.py b/tests/test_connections.py
index e3c84de..b63bded 100644
--- a/tests/test_connections.py
+++ b/tests/test_connections.py
@@ -330,8 +330,79 @@ def test_none_values_excluded(self):
         assert "port" not in payload
         assert "username" not in payload
         assert "password" not in payload
+        assert "authentication_type" not in payload
         assert payload == {"type": "postgresql"}
 
+    def test_s3_iam_role_nests_under_parameters(self):
+        payload = build_create_connection_payload(
+            "s3",
+            name="s3-iam",
+            uri="s3://my-bucket",
+            authentication_type="IAM_ROLE",
+            role_arn="arn:aws:iam::123:role/MyRole",
+            external_id="ext-123",
+        )
+        # IAM fields must be nested in parameters (controlplane uses
+        # map_to="parameters" for them; top-level rejected with 422).
+        assert payload["parameters"] == {
+            "authentication_type": "IAM_ROLE",
+            "role_arn": "arn:aws:iam::123:role/MyRole",
+            "external_id": "ext-123",
+        }
+        # And NOT at top level.
+        assert "authentication_type" not in payload
+        assert "role_arn" not in payload
+        assert "external_id" not in payload
+        assert "access_key" not in payload
+        assert "secret_key" not in payload
+
+    def test_athena_iam_role_nests_under_parameters(self):
+        payload = build_create_connection_payload(
+            "athena",
+            name="athena-iam",
+            authentication_type="IAM_ROLE",
+            role_arn="arn:aws:iam::123:role/AthenaRole",
+        )
+        assert payload["type"] == "athena"
+        assert payload["parameters"] == {
+            "authentication_type": "IAM_ROLE",
+            "role_arn": "arn:aws:iam::123:role/AthenaRole",
+        }
+
+    def test_redshift_iam_role_nests_under_parameters(self):
+        payload = build_create_connection_payload(
+            "redshift",
+            name="rs-iam",
+            host="cluster.redshift.amazonaws.com",
+            authentication_type="IAM_ROLE",
+            role_arn="arn:aws:iam::123:role/RedshiftRole",
+        )
+        assert payload["type"] == "redshift"
+        assert payload["parameters"]["authentication_type"] == "IAM_ROLE"
+        assert payload["parameters"]["role_arn"] == "arn:aws:iam::123:role/RedshiftRole"
+
+    def test_iam_role_merges_with_existing_parameters(self):
+        payload = build_create_connection_payload(
+            "s3",
+            name="s3-iam",
+            uri="s3://my-bucket",
+            authentication_type="IAM_ROLE",
+            role_arn="arn:aws:iam::123:role/MyRole",
+            parameters={"some_extra": "value"},
+        )
+        # The catch-all --parameters merge happens after IAM nesting, so
+        # both end up represented (catch-all is top-level, IAM is nested).
+        assert payload["parameters"]["authentication_type"] == "IAM_ROLE"
+        assert payload["some_extra"] == "value"
+
+    def test_iam_role_without_role_arn_rejected(self):
+        with pytest.raises(ValueError, match="--role-arn is required"):
+            build_create_connection_payload(
+                "s3",
+                name="bad",
+                authentication_type="IAM_ROLE",
+            )
+
 
 class TestBuildUpdateConnectionPayload:
     def test_includes_non_none_values(self):
@@ -346,6 +417,36 @@ def test_empty_when_all_none(self):
         payload = build_update_connection_payload(name=None, host=None)
         assert payload == {}
 
+    def test_iam_role_fields_nested_under_parameters(self):
+        payload = build_update_connection_payload(
+            authentication_type="IAM_ROLE",
+            role_arn="arn:aws:iam::123:role/MyRole",
+            external_id="ext-1",
+        )
+        assert payload == {
+            "parameters": {
+                "authentication_type": "IAM_ROLE",
+                "role_arn": "arn:aws:iam::123:role/MyRole",
+                "external_id": "ext-1",
+            }
+        }
+
+    def test_update_mixes_top_level_and_iam_fields(self):
+        payload = build_update_connection_payload(
+            name="renamed",
+            authentication_type="IAM_ROLE",
+            role_arn="arn:aws:iam::123:role/MyRole",
+        )
+        assert payload["name"] == "renamed"
+        assert payload["parameters"] == {
+            "authentication_type": "IAM_ROLE",
+            "role_arn": "arn:aws:iam::123:role/MyRole",
+        }
+
+    def test_iam_role_without_role_arn_rejected_on_update(self):
+        with pytest.raises(ValueError, match="--role-arn is required"):
+            build_update_connection_payload(authentication_type="IAM_ROLE")
+
 
 # ══════════════════════════════════════════════════════════════════════════
 # 3. SECRETS UTILITIES
@@ -476,6 +577,59 @@ def test_create_inline(self, mock_gc, mock_create, cli_runner):
         # Verify sensitive data is redacted in output
         assert "secret" not in result.output or "redacted" in result.output
 
+    @patch("qualytics.cli.connections.create_connection")
+    @patch("qualytics.cli.connections.get_client")
+    def test_create_s3_iam_role(self, mock_gc, mock_create, cli_runner):
+        mock_gc.return_value = _mock_client()
+        mock_create.return_value = {"id": 9, "name": "s3-iam", "type": "s3"}
+        result = cli_runner.invoke(
+            app,
+            [
+                "connections",
+                "create",
+                "--type",
+                "s3",
+                "--name",
+                "s3-iam",
+                "--uri",
+                "s3://my-bucket",
+                "--authentication-type",
+                "IAM_ROLE",
+                "--role-arn",
+                "arn:aws:iam::123:role/MyRole",
+                "--external-id",
+                "ext-1",
+            ],
+        )
+        assert result.exit_code == 0
+        payload = mock_create.call_args.args[1]
+        assert payload["parameters"] == {
+            "authentication_type": "IAM_ROLE",
+            "role_arn": "arn:aws:iam::123:role/MyRole",
+            "external_id": "ext-1",
+        }
+        assert "authentication_type" not in payload
+        assert "role_arn" not in payload
+
+    @patch("qualytics.cli.connections.get_client")
+    def test_create_iam_role_without_role_arn_fails(self, mock_gc, cli_runner):
+        mock_gc.return_value = _mock_client()
+        result = cli_runner.invoke(
+            app,
+            [
+                "connections",
+                "create",
+                "--type",
+                "s3",
+                "--name",
+                "bad",
+                "--authentication-type",
+                "IAM_ROLE",
+            ],
+        )
+        assert result.exit_code == 1
+        assert "--role-arn is required" in result.output
+
     @patch("qualytics.cli.connections.get_client")
     def test_create_dry_run(self, mock_gc, cli_runner):
         mock_gc.return_value = _mock_client()
@@ -631,6 +785,29 @@ def test_update_partial(self, mock_gc, mock_update, cli_runner):
         assert payload.get("host") == "new-host"
         assert "name" not in payload
 
+    @patch("qualytics.cli.connections.update_connection")
+    @patch("qualytics.cli.connections.get_client")
+    def test_update_iam_role(self, mock_gc, mock_update, cli_runner):
+        mock_gc.return_value = _mock_client()
+        mock_update.return_value = {"id": 1, "authentication_type": "IAM_ROLE"}
+        result = cli_runner.invoke(
+            app,
+            [
+                "connections",
+                "update",
+                "--id",
+                "1",
+                "--authentication-type",
+                "IAM_ROLE",
+                "--role-arn",
+                "arn:aws:iam::123:role/NewRole",
+            ],
+        )
+        assert result.exit_code == 0
+        payload = mock_update.call_args[0][2]
+        assert payload["parameters"]["authentication_type"] == "IAM_ROLE"
+        assert payload["parameters"]["role_arn"] == "arn:aws:iam::123:role/NewRole"
+
     @patch("qualytics.cli.connections.get_client")
     def test_update_no_fields_fails(self, mock_gc, cli_runner):
         mock_gc.return_value = _mock_client()
diff --git a/tests/test_operations.py b/tests/test_operations.py
index d8a12ef..997930f 100644
--- a/tests/test_operations.py
+++ b/tests/test_operations.py
@@ -273,7 +273,7 @@ def test_profile_builds_correct_payload(self, mock_wait, mock_run):
             [42],
             ["orders"],
             None,
-            3,
+            "high",
             True,
             None,
             None,
@@ -290,7 +290,48 @@ def test_profile_builds_correct_payload(self, mock_wait, mock_run):
         assert payload["type"] == "profile"
         assert payload["datastore_id"] == 42
         assert payload["container_names"] == ["orders"]
-        assert payload["inference_threshold"] == 3
+        assert payload["ai_effort"] == "high"
+
+    def test_ai_effort_numeric_strings_are_normalized(self):
+        from qualytics.services.operations import _normalize_ai_effort
+
+        assert _normalize_ai_effort("0") == "off"
+        assert _normalize_ai_effort("1") == "low"
+        assert _normalize_ai_effort("2") == "medium"
+        assert _normalize_ai_effort("3") == "high"
+        assert _normalize_ai_effort("4") == "xhigh"
+        assert _normalize_ai_effort("5") == "max"
+        assert _normalize_ai_effort("high") == "high"
+        assert _normalize_ai_effort(None) is None
+
+    @patch("qualytics.services.operations.run_operation")
+    @patch("qualytics.services.operations.wait_for_operation")
+    def test_profile_legacy_numeric_ai_effort_is_normalized(self, mock_wait, mock_run):
+        mock_run.return_value = {"id": 201}
+        mock_wait.return_value = {"result": "success", "message": None}
+
+        from qualytics.services.operations import run_profile
+
+        run_profile(
+            _mock_client(),
+            [42],
+            None,
+            None,
+            "3",  # legacy numeric string
+            None,
+            None,
+            None,
+            None,
+            None,
+            None,
+            None,
+            None,
+            False,
+            poll_interval=1,
+            timeout=10,
+        )
+        payload = mock_run.call_args.args[1]
+        assert payload["ai_effort"] == "high"
 
     @patch("qualytics.services.operations.run_operation")
     @patch("qualytics.services.operations.wait_for_operation")
@@ -321,6 +362,64 @@ def test_scan_builds_correct_payload(self, mock_wait, mock_run):
         assert payload["incremental"] is True
         assert payload["remediation"] == "append"
         assert payload["enrichment_source_record_limit"] == 100
+        # Default: not in payload, so server default (True) fires.
+        assert "auto_resolve_passed_anomalies" not in payload
+
+    @patch("qualytics.services.operations.run_operation")
+    @patch("qualytics.services.operations.wait_for_operation")
+    def test_scan_auto_resolve_passed_anomalies_true(self, mock_wait, mock_run):
+        mock_run.return_value = {"id": 301}
+        mock_wait.return_value = {"result": "success", "message": None}
+
+        from qualytics.services.operations import run_scan
+
+        client = _mock_client()
+        run_scan(
+            client,
+            [42],
+            None,
+            None,
+            None,
+            "none",
+            None,
+            None,
+            None,
+            None,
+            False,
+            auto_resolve_passed_anomalies=True,
+            poll_interval=1,
+            timeout=10,
+        )
+        payload = mock_run.call_args.args[1]
+        assert payload["auto_resolve_passed_anomalies"] is True
+
+    @patch("qualytics.services.operations.run_operation")
+    @patch("qualytics.services.operations.wait_for_operation")
+    def test_scan_auto_resolve_passed_anomalies_false(self, mock_wait, mock_run):
+        mock_run.return_value = {"id": 302}
+        mock_wait.return_value = {"result": "success", "message": None}
+
+        from qualytics.services.operations import run_scan
+
+        client = _mock_client()
+        run_scan(
+            client,
+            [42],
+            None,
+            None,
+            None,
+            "none",
+            None,
+            None,
+            None,
+            None,
+            False,
+            auto_resolve_passed_anomalies=False,
+            poll_interval=1,
+            timeout=10,
+        )
+        payload = mock_run.call_args.args[1]
+        assert payload["auto_resolve_passed_anomalies"] is False
 
     @patch("qualytics.services.operations.run_operation")
     @patch("qualytics.services.operations.wait_for_operation")
@@ -454,8 +553,8 @@ def test_profile_with_kebab_case_flags(self, mock_run, mock_get_client, cli_runn
                 "orders,customers",
                 "--container-tags",
                 "production",
-                "--inference-threshold",
-                "3",
+                "--ai-effort",
+                "high",
                 "--infer-as-draft",
                 "--max-records-analyzed-per-partition",
                 "1000",
@@ -468,9 +567,80 @@ def test_profile_with_kebab_case_flags(self, mock_run, mock_get_client, cli_runn
         kwargs = mock_run.call_args.kwargs
         assert kwargs["container_names"] == ["orders", "customers"]
         assert kwargs["container_tags"] == ["production"]
-        assert kwargs["inference_threshold"] == 3
+        assert kwargs["ai_effort"] == "high"
         assert kwargs["max_records_analyzed_per_partition"] == 1000
 
+    @patch("qualytics.cli.operations.get_client")
+    @patch("qualytics.cli.operations.run_profile")
+    def test_profile_deprecated_inference_threshold_flag(
+        self, mock_run, mock_get_client, cli_runner
+    ):
+        mock_get_client.return_value = _mock_client()
+        result = cli_runner.invoke(
+            app,
+            [
+                "operations",
+                "profile",
+                "--datastore-id",
+                "42",
+                "--inference-threshold",
+                "3",
+            ],
+        )
+        assert result.exit_code == 0
+        assert "deprecated" in result.output.lower()
+        assert mock_run.call_args.kwargs["ai_effort"] == "3"
+
+    @patch("qualytics.cli.operations.get_client")
+    @patch("qualytics.cli.operations.run_profile")
+    def test_ai_effort_wins_over_inference_threshold(
+        self, mock_run, mock_get_client, cli_runner
+    ):
+        mock_get_client.return_value = _mock_client()
+        result = cli_runner.invoke(
+            app,
+            [
+                "operations",
+                "profile",
+                "--datastore-id",
+                "42",
+                "--ai-effort",
+                "max",
+                "--inference-threshold",
+                "1",
+            ],
+        )
+        assert result.exit_code == 0
+        assert mock_run.call_args.kwargs["ai_effort"] == "max"
+
+    @patch("qualytics.cli.operations.get_client")
+    @patch("qualytics.cli.operations.run_profile")
+    def test_ai_effort_is_case_insensitive(self, mock_run, mock_get_client, cli_runner):
+        mock_get_client.return_value = _mock_client()
+        result = cli_runner.invoke(
+            app,
+            ["operations", "profile", "--datastore-id", "42", "--ai-effort", "High"],
+        )
+        assert result.exit_code == 0
+        assert mock_run.call_args.kwargs["ai_effort"] == "high"
+
+    @patch("qualytics.cli.operations.get_client")
+    def test_rejects_invalid_ai_effort(self, mock_get_client, cli_runner):
+        mock_get_client.return_value = _mock_client()
+        result = cli_runner.invoke(
+            app,
+            [
+                "operations",
+                "profile",
+                "--datastore-id",
+                "42",
+                "--ai-effort",
+                "10",
+            ],
+        )
+        assert result.exit_code == 1
+        assert "must be one of" in result.output
+
     @patch("qualytics.cli.operations.get_client")
     def test_rejects_invalid_max_records(self, mock_get_client, cli_runner):
         mock_get_client.return_value = _mock_client()
@@ -529,6 +699,44 @@ def test_scan_with_kebab_case_flags(self, mock_run, mock_get_client, cli_runner)
         kwargs = mock_run.call_args.kwargs
         assert kwargs["remediation"] == "append"
         assert kwargs["enrichment_source_record_limit"] == 500
+        # No flag passed → None, so we don't override the server default.
+        assert kwargs["auto_resolve_passed_anomalies"] is None
+
+    @patch("qualytics.cli.operations.get_client")
+    @patch("qualytics.cli.operations.run_scan")
+    def test_scan_with_auto_resolve_on(self, mock_run, mock_get_client, cli_runner):
+        mock_get_client.return_value = _mock_client()
+        result = cli_runner.invoke(
+            app,
+            [
+                "operations",
+                "scan",
+                "--datastore-id",
+                "42",
+                "--auto-resolve-passed-anomalies",
+                "--background",
+            ],
+        )
+        assert result.exit_code == 0
+        assert mock_run.call_args.kwargs["auto_resolve_passed_anomalies"] is True
+
+    @patch("qualytics.cli.operations.get_client")
+    @patch("qualytics.cli.operations.run_scan")
+    def test_scan_with_auto_resolve_off(self, mock_run, mock_get_client, cli_runner):
+        mock_get_client.return_value = _mock_client()
+        result = cli_runner.invoke(
+            app,
+            [
+                "operations",
+                "scan",
+                "--datastore-id",
+                "42",
+                "--no-auto-resolve-passed-anomalies",
+                "--background",
+            ],
+        )
+        assert result.exit_code == 0
+        assert mock_run.call_args.kwargs["auto_resolve_passed_anomalies"] is False
 
     @patch("qualytics.cli.operations.get_client")
     def test_rejects_invalid_remediation(self, mock_get_client, cli_runner):
diff --git a/tests/test_quality_checks.py b/tests/test_quality_checks.py
index 9d73811..c0082a0 100644
--- a/tests/test_quality_checks.py
+++ b/tests/test_quality_checks.py
@@ -328,6 +328,47 @@ def test_create_payload_defaults_missing_fields(self):
         assert payload["properties"] == {}
         assert payload["tags"] == []
         assert payload["status"] == "Active"
+        assert "owner_id" not in payload
+        assert "default_anomaly_assignee_id" not in payload
+
+    def test_create_payload_includes_ownership_when_present(self):
+        check = {
+            "rule_type": "notNull",
+            "owner_id": 7,
+            "default_anomaly_assignee_id": 12,
+        }
+        payload = _build_create_payload(check, container_id=1)
+        assert payload["owner_id"] == 7
+        assert payload["default_anomaly_assignee_id"] == 12
+
+    def test_update_payload_includes_ownership_when_present(self):
+        check = {
+            "description": "x",
+            "owner_id": 9,
+            "default_anomaly_assignee_id": 13,
+        }
+        payload = _build_update_payload(check)
+        assert payload["owner_id"] == 9
+        assert payload["default_anomaly_assignee_id"] == 13
+
+    def test_payload_treats_zero_as_clear(self):
+        # YAML `owner_id: 0` and `--owner-id 0` should behave identically:
+        # both mean "clear the value" (user IDs are positive).
+        check = {
+            "description": "x",
+            "owner_id": 0,
+            "default_anomaly_assignee_id": 0,
+        }
+        update_payload = _build_update_payload(check)
+        assert update_payload["owner_id"] is None
+        assert update_payload["default_anomaly_assignee_id"] is None
+
+        create_payload = _build_create_payload(
+            {"rule_type": "notNull", "owner_id": 0, "default_anomaly_assignee_id": 0},
+            container_id=1,
+        )
+        assert create_payload["owner_id"] is None
+        assert create_payload["default_anomaly_assignee_id"] is None
 
 
 # ── Load from directory ──────────────────────────────────────────────────
@@ -920,6 +961,121 @@ def test_update(self, mock_gc, mock_update, cli_runner, tmp_path):
         assert result.exit_code == 0
         assert "updated successfully" in result.output
 
+    @patch("qualytics.cli.checks.update_quality_check")
+    @patch("qualytics.cli.checks.get_client")
+    def test_update_with_owner_flag(self, mock_gc, mock_update, cli_runner, tmp_path):
+        mock_gc.return_value = _mock_client()
+        mock_update.return_value = {"id": 42}
+
+        check_file = tmp_path / "check.yaml"
+        check_file.write_text("description: Updated\nfields: [email]\n")
+
+        result = cli_runner.invoke(
+            app,
+            [
+                "checks",
+                "update",
+                "--id",
+                "42",
+                "--file",
+                str(check_file),
+                "--owner-id",
+                "7",
+                "--default-anomaly-assignee-id",
+                "12",
+            ],
+        )
+        assert result.exit_code == 0
+        payload = mock_update.call_args.args[2]
+        assert payload["owner_id"] == 7
+        assert payload["default_anomaly_assignee_id"] == 12
+
+    @patch("qualytics.cli.checks.update_quality_check")
+    @patch("qualytics.cli.checks.get_client")
+    def test_update_owner_zero_clears(self, mock_gc, mock_update, cli_runner, tmp_path):
+        mock_gc.return_value = _mock_client()
+        mock_update.return_value = {"id": 42}
+
+        check_file = tmp_path / "check.yaml"
+        check_file.write_text("description: x\nowner_id: 99\n")
+
+        result = cli_runner.invoke(
+            app,
+            [
+                "checks",
+                "update",
+                "--id",
+                "42",
+                "--file",
+                str(check_file),
+                "--owner-id",
+                "0",
+            ],
+        )
+        assert result.exit_code == 0
+        payload = mock_update.call_args.args[2]
+        assert payload["owner_id"] is None
+
+    @patch("qualytics.cli.checks.update_quality_check")
+    @patch("qualytics.cli.checks.get_client")
+    def test_update_assignee_zero_clears(
+        self, mock_gc, mock_update, cli_runner, tmp_path
+    ):
+        mock_gc.return_value = _mock_client()
+        mock_update.return_value = {"id": 42}
+
+        check_file = tmp_path / "check.yaml"
+        check_file.write_text("description: x\ndefault_anomaly_assignee_id: 50\n")
+
+        result = cli_runner.invoke(
+            app,
+            [
+                "checks",
+                "update",
+                "--id",
+                "42",
+                "--file",
+                str(check_file),
+                "--default-anomaly-assignee-id",
+                "0",
+            ],
+        )
+        assert result.exit_code == 0
+        payload = mock_update.call_args.args[2]
+        assert payload["default_anomaly_assignee_id"] is None
+
+    @patch("qualytics.cli.checks.create_quality_check")
+    @patch("qualytics.cli.checks.get_table_ids")
+    @patch("qualytics.cli.checks.get_client")
+    def test_create_with_owner_flag_overrides(
+        self, mock_gc, mock_tables, mock_create, cli_runner, tmp_path
+    ):
+        mock_gc.return_value = _mock_client()
+        mock_tables.return_value = {"orders": 100}
+        mock_create.return_value = {"id": 999}
+
+        check_file = tmp_path / "check.yaml"
+        check_file.write_text(
+            "rule_type: notNull\ncontainer: orders\nfields: [order_id]\nowner_id: 1\n"
+        )
+
+        result = cli_runner.invoke(
+            app,
+            [
+                "checks",
+                "create",
+                "--datastore-id",
+                "42",
+                "--file",
+                str(check_file),
+                "--owner-id",
+                "5",
+            ],
+        )
+        assert result.exit_code == 0
+        payload = mock_create.call_args.args[1]
+        assert payload["owner_id"] == 5
+
 
 class TestChecksDeleteCLI:
     @patch("qualytics.cli.checks.delete_quality_check")