diff --git a/docs/release-notes/2.5.0.md b/docs/release-notes/2.5.0.md
index c8a62a9..ffa5813 100644
--- a/docs/release-notes/2.5.0.md
+++ b/docs/release-notes/2.5.0.md
@@ -62,6 +62,8 @@ Every asset carries a GitHub build-provenance attestation. Linux binaries are st
 
 ## Intentional differences from 2.x
 
+- **Connection URLs:** runtime base URLs containing user information, queries, fragments or whitespace are refused before a request. Use token options or profile token references for authentication.
+- **Doctor output:** human diagnostics show untrusted formatter markup literally and escape terminal controls and embedded line breaks. Machine JSON retains Server data as structurally encoded values.
 - **Shell completion:** generating a script requires an executable name made of ASCII letters, digits, dots, underscores and hyphens, starting with a letter or underscore. Unsafe names are refused before any shell code is emitted. Completion debug logs now use a private `completion` directory alongside the CLI config file, and refuse unsafe log entries instead of following temporary-file links.
 - **Prompting without a terminal:** 2.x treated a non-terminal stdin as interactive and printed a "Do you want to run … instead?" prompt to stdout before exiting 1. 2.5.0 prompts only on a terminal; otherwise it reports the error on stderr, as 2.x did with `-n`.
 - **`--format=xml`:** this works in the standalone binary. The 2.x standalone binary failed with a PHP fatal error, because it lacked the DOM extension. The output matches the 2.x PHAR.
diff --git a/rust/README.md b/rust/README.md
index 40b6a34..2b364d0 100644
--- a/rust/README.md
+++ b/rust/README.md
@@ -21,6 +21,7 @@ every command group reaches parity.
 | `env:list`, `env:set`, `env:use`, `env:show`, `env:delete` | Ported |
 | `server:health`, `server:info` (and `cluster:info`) | Ported |
 | Compatibility preflight, session and `--version` warnings, exit 8 refusals | Ported |
+| `doctor`: redacted connection sources, server metadata, warnings and next steps | Ported |
 | Workflow, namespace and schedule commands | Next slices |
 | `--tls-verify=false` against HTTPS | Refused until durable-workflow/sdk-rust#180 |
 
diff --git a/rust/parity/cases.json b/rust/parity/cases.json
index abcec34..66256dc 100644
--- a/rust/parity/cases.json
+++ b/rust/parity/cases.json
@@ -2464,6 +2464,619 @@
     "-a0.9",
     "-idw"
    ]
+  },
+  {
+   "name": "doctor JSON",
+   "argv": [
+    "doctor",
+    "--output=json",
+    "--server={mock}"
+   ],
+   "mock": []
+  },
+  {
+   "name": "doctor JSONL",
+   "argv": [
+    "doctor",
+    "--output=jsonl",
+    "--server={mock}"
+   ],
+   "mock": []
+  },
+  {
+   "name": "doctor human",
+   "argv": [
+    "doctor",
+    "--server={mock}"
+   ],
+   "mock": []
+  },
+  {
+   "name": "doctor ANSI",
+   "argv": [
+    "doctor",
+    "--ansi",
+    "--server={mock}"
+   ],
+   "mock": []
+  },
+  {
+   "name": "doctor quiet",
+   "argv": [
+    "doctor",
+    "-q",
+    "--server={mock}"
+   ],
+   "mock": []
+  },
+  {
+   "name": "doctor clean with flag token",
+   "argv": [
+    "doctor",
+    "--token=parity-test-token",
+    "--output=json",
+    "--server={mock}"
+   ],
+   "mock": []
+  },
+  {
+   "name": "doctor environment connection",
+   "argv": [
+    "doctor",
+    "--output=json"
+   ],
+   "mock": [],
+   "env": {
+    "DURABLE_WORKFLOW_SERVER_URL": "{mock}",
+    "DURABLE_WORKFLOW_NAMESPACE": "env-namespace",
+    "DURABLE_WORKFLOW_AUTH_TOKEN": "parity-test-token",
+    "DURABLE_WORKFLOW_TLS_VERIFY": "false"
+   }
+  },
+  {
+   "name": "doctor flag profile",
+   "argv": [
+    "doctor",
+    "--env=prod",
+    "--output=json"
+   ],
+   "mock": [],
+   "env": {},
+   "setup": [
+    [
+     "env:set",
+     "prod",
+     "--server={mock}",
+     "--namespace=profile-ns",
+     "--token=profile-test-token",
+     "--tls-verify=false"
+    ]
+   ]
+  },
+  {
+   "name": "doctor selected profile",
+   "argv": [
+    "doctor",
+    "--output=json"
+   ],
+   "mock": [],
+   "env": {},
+   "setup": [
+    [
+     "env:set",
+     "prod",
+     "--server={mock}",
+     "--namespace=profile-ns",
+     "--token=profile-test-token",
+     "--tls-verify=false"
+    ],
+    [
+     "env:use",
+     "prod"
+    ]
+   ]
+  },
+  {
+   "name": "doctor DW_ENV profile",
+   "argv": [
+    "doctor",
+    "--output=json"
+   ],
+   "mock": [],
+   "env": {
+    "DW_ENV": "prod"
+   },
+   "setup": [
+    [
+     "env:set",
+     "prod",
+     "--server={mock}",
+     "--namespace=profile-ns",
+     "--token=profile-test-token",
+     "--tls-verify=false"
+    ]
+   ]
+  },
+  {
+   "name": "doctor profile env token present",
+   "argv": [
+    "doctor",
+    "--env=prod",
+    "--output=json"
+   ],
+   "mock": [],
+   "env": {
+    "PARITY_DOCTOR_TOKEN": "profile-env-test-token"
+   },
+   "setup": [
+    [
+     "env:set",
+     "prod",
+     "--server={mock}",
+     "--token-env=PARITY_DOCTOR_TOKEN"
+    ]
+   ]
+  },
+  {
+   "name": "doctor profile env token missing",
+   "argv": [
+    "doctor",
+    "--env=prod",
+    "--output=json"
+   ],
+   "mock": [],
+   "env": {
+    "PARITY_DOCTOR_TOKEN": ""
+   },
+   "setup": [
+    [
+     "env:set",
+     "prod",
+     "--server={mock}",
+     "--token-env=PARITY_DOCTOR_TOKEN"
+    ]
+   ]
+  },
+  {
+   "name": "doctor flag precedence",
+   "argv": [
+    "doctor",
+    "--env=prod",
+    "--namespace=flag-ns",
+    "--token=flag-test-token",
+    "--tls-verify=true",
+    "--output=json"
+   ],
+   "mock": [],
+   "env": {
+    "DURABLE_WORKFLOW_NAMESPACE": "env-ns",
+    "DURABLE_WORKFLOW_AUTH_TOKEN": "env-test-token",
+    "DURABLE_WORKFLOW_TLS_VERIFY": "false"
+   },
+   "setup": [
+    [
+     "env:set",
+     "prod",
+     "--server={mock}",
+     "--namespace=profile-ns",
+     "--token=profile-test-token",
+     "--tls-verify=false"
+    ]
+   ]
+  },
+  {
+   "name": "doctor TLS disabled",
+   "argv": [
+    "doctor",
+    "--tls-verify=false",
+    "--output=json",
+    "--server={mock}"
+   ],
+   "mock": []
+  },
+  {
+   "name": "doctor incompatible control plane",
+   "argv": [
+    "doctor",
+    "--output=json",
+    "--server={mock}"
+   ],
+   "mock": [
+    [
+     [
+      "control_plane",
+      "version"
+     ],
+     "3"
+    ]
+   ]
+  },
+  {
+   "name": "doctor incompatible worker protocol",
+   "argv": [
+    "doctor",
+    "--output=json",
+    "--server={mock}"
+   ],
+   "mock": [
+    [
+     [
+      "worker_protocol",
+      "version"
+     ],
+     "99.0"
+    ]
+   ]
+  },
+  {
+   "name": "doctor missing protocol metadata",
+   "argv": [
+    "doctor",
+    "--output=json",
+    "--server={mock}"
+   ],
+   "mock": [
+    [
+     [
+      "control_plane"
+     ],
+     null
+    ],
+    [
+     [
+      "worker_protocol"
+     ],
+     null
+    ],
+    [
+     [
+      "auth_composition_contract"
+     ],
+     null
+    ]
+   ]
+  },
+  {
+   "name": "doctor malformed protocol metadata",
+   "argv": [
+    "doctor",
+    "--output=json",
+    "--server={mock}"
+   ],
+   "mock": [
+    [
+     [
+      "version"
+     ],
+     false
+    ],
+    [
+     [
+      "server_id"
+     ],
+     {}
+    ],
+    [
+     [
+      "auth_composition_contract"
+     ],
+     []
+    ],
+    [
+     [
+      "control_plane"
+     ],
+     []
+    ],
+    [
+     [
+      "worker_protocol"
+     ],
+     {
+      "version": 7,
+      "invocable_carrier_contract": {
+       "version": true,
+       "scope": {
+        "task_kinds": {
+         "x": "activity_task"
+        }
+       }
+      },
+      "external_task_input_contract": {
+       "version": "2x",
+       "fixtures": {
+        "unknown": {},
+        "activity_task": {}
+       }
+      }
+     }
+    ]
+   ]
+  },
+  {
+   "name": "doctor native published metadata",
+   "argv": [
+    "doctor",
+    "--output=json",
+    "--server={mock}"
+   ],
+   "mock": [],
+   "mock_routes": {
+    "/api/cluster/info": {
+     "body": {
+      "server_id": "9b667a0d792b",
+      "version": "3.0.5",
+      "implementation": "rust",
+      "control_plane": {
+       "version": "2",
+       "request_contract": {
+        "schema": "durable-workflow.v2.control-plane-request.contract",
+        "version": 1,
+        "operations": {
+         "list": {
+          "fields": {
+           "status": {
+            "canonical_values": [
+             "running",
+             "completed",
+             "failed"
+            ],
+            "rejected_aliases": {
+             "cancelled": "failed",
+             "terminated": "failed",
+             "pending": "running",
+             "waiting": "running"
+            }
+           }
+          }
+         },
+         "update": {
+          "fields": {
+           "wait_for": {
+            "canonical_values": [
+             "accepted",
+             "completed"
+            ]
+           },
+           "wait_timeout_seconds": {
+            "minimum": 1,
+            "maximum": 60
+           }
+          }
+         }
+        }
+       }
+      },
+      "worker_protocol": {
+       "version": "1.20"
+      },
+      "auth_composition_contract": {
+       "schema": "durable-workflow.v2.auth-composition.contract",
+       "version": 1,
+       "scope": "external_execution_carriers",
+       "unknown_field_policy": "ignore_additive_reject_unknown_required",
+       "precedence": {
+        "connection_values": [
+         "flag",
+         "environment",
+         "selected_profile",
+         "default"
+        ],
+        "profile_selection": [
+         "flag_env",
+         "DW_ENV",
+         "current_profile",
+         "default_profile"
+        ],
+        "notes": [
+         "flags win only for the current invocation",
+         "environment values are portable carrier inputs and must not be persisted by default",
+         "selected profile values are persisted configuration, but secret fields should be references",
+         "defaults are explicit and must appear in effective-config diagnostics"
+        ]
+       },
+       "canonical_environment": {
+        "server_url": "DURABLE_WORKFLOW_SERVER_URL",
+        "namespace": "DURABLE_WORKFLOW_NAMESPACE",
+        "auth_token": "DURABLE_WORKFLOW_AUTH_TOKEN",
+        "tls_verify": "DURABLE_WORKFLOW_TLS_VERIFY",
+        "profile": "DW_ENV"
+       },
+       "auth_material": {
+        "token": {
+         "status": "supported",
+         "transport": "bearer_authorization_header",
+         "persisted_as": "secret_reference_or_profile_env_reference",
+         "effective_config_value": "redacted"
+        },
+        "mtls": {
+         "status": "reserved",
+         "persisted_as": "certificate_and_key_references",
+         "effective_config_value": "references_only"
+        },
+        "signed_headers": {
+         "status": "reserved",
+         "persisted_as": "key_reference_and_header_allowlist",
+         "effective_config_value": "references_only"
+        },
+        "extensions": {
+         "status": "reserved",
+         "rule": "new auth material types must define redaction, persistence, and source precedence before use"
+        }
+       },
+       "effective_config": {
+        "required_fields": [
+         "server_url",
+         "namespace",
+         "profile",
+         "auth",
+         "tls",
+         "identity"
+        ],
+        "source_values": [
+         "flag",
+         "environment",
+         "selected_profile",
+         "profile_env",
+         "default",
+         "server"
+        ],
+        "field_contracts": {
+         "server_url": {
+          "value": "normalized_absolute_url",
+          "source": "one_of_source_values"
+         },
+         "namespace": {
+          "value": "non_empty_namespace_name",
+          "source": "one_of_source_values"
+         },
+         "profile": {
+          "name": "nullable_string",
+          "source": "flag_env_or_DW_ENV_or_current_profile_or_default_profile"
+         },
+         "auth": {
+          "type": "token_or_mtls_or_signed_headers_or_none",
+          "source": "one_of_source_values",
+          "value": "redacted_or_reference_only"
+         },
+         "tls": {
+          "verify": "boolean",
+          "source": "one_of_source_values"
+         },
+         "identity": {
+          "subject": "nullable_string",
+          "roles": "list_of_strings",
+          "source": "server_or_carrier"
+         }
+        }
+       },
+       "redaction": {
+        "never_echo": [
+         "bearer_tokens",
+         "private_keys",
+         "shared_signature_keys",
+         "client_certificate_private_key_material",
+         "raw_authorization_headers"
+        ],
+        "allowed_diagnostics": [
+         "redacted",
+         "secret_reference_name",
+         "environment_variable_name",
+         "profile_name",
+         "certificate_reference_name",
+         "key_reference_name"
+        ]
+       },
+       "resolution": {
+        "server_url": "carriers normalize and use the winning absolute URL without appending hidden path defaults",
+        "namespace": "carriers send the winning namespace explicitly when the protocol supports namespace headers",
+        "identity": "servers may report authenticated subject and roles; carriers must treat identity as server-asserted when present",
+        "tls": "carriers must fail closed on invalid TLS unless the winning tls.verify value is false"
+       }
+      }
+     }
+    }
+   }
+  },
+  {
+   "name": "doctor HTTP 401",
+   "argv": [
+    "doctor",
+    "--output=json",
+    "--server={mock}"
+   ],
+   "mock": [],
+   "mock_routes": {
+    "/api/cluster/info": {
+     "status": 401,
+     "body": {
+      "message": "Synthetic diagnostic refusal"
+     }
+    }
+   }
+  },
+  {
+   "name": "doctor HTTP 403",
+   "argv": [
+    "doctor",
+    "--output=json",
+    "--server={mock}"
+   ],
+   "mock": [],
+   "mock_routes": {
+    "/api/cluster/info": {
+     "status": 403,
+     "body": {
+      "message": "Synthetic diagnostic refusal"
+     }
+    }
+   }
+  },
+  {
+   "name": "doctor HTTP 404",
+   "argv": [
+    "doctor",
+    "--output=json",
+    "--server={mock}"
+   ],
+   "mock": [],
+   "mock_routes": {
+    "/api/cluster/info": {
+     "status": 404,
+     "body": {
+      "message": "Synthetic diagnostic refusal"
+     }
+    }
+   }
+  },
+  {
+   "name": "doctor HTTP 422",
+   "argv": [
+    "doctor",
+    "--output=json",
+    "--server={mock}"
+   ],
+   "mock": [],
+   "mock_routes": {
+    "/api/cluster/info": {
+     "status": 422,
+     "body": {
+      "message": "Synthetic diagnostic refusal"
+     }
+    }
+   }
+  },
+  {
+   "name": "doctor HTTP 500",
+   "argv": [
+    "doctor",
+    "--output=json",
+    "--server={mock}"
+   ],
+   "mock": [],
+   "mock_routes": {
+    "/api/cluster/info": {
+     "status": 500,
+     "body": {
+      "message": "Synthetic diagnostic refusal"
+     }
+    }
+   }
+  },
+  {
+   "name": "doctor unreachable JSON",
+   "argv": [
+    "doctor",
+    "--server=http://127.0.0.1:9",
+    "--output=json"
+   ]
+  },
+  {
+   "name": "doctor unreachable human",
+   "argv": [
+    "doctor",
+    "--server=http://127.0.0.1:9"
+   ]
   }
  ]
 }
diff --git a/rust/parity/parity.py b/rust/parity/parity.py
index a6af970..af87486 100755
--- a/rust/parity/parity.py
+++ b/rust/parity/parity.py
@@ -49,6 +49,7 @@
 BUILD_IDENTITY = re.compile(r"\(commit [^,]+, built [^)]+\)")
 MARKDOWN_TITLE = re.compile(r"^(dw <version>)\n=+$", re.M)
 TRANSPORT_LINE = re.compile(r"^(Server unreachable|Request timed out): .*$", re.M)
+DOCTOR_TRANSPORT_LINE = re.compile(r"^(  Error: )(Server unreachable|Request timed out): .*$", re.M)
 LOOPBACK_PORT = re.compile(r"127\.0\.0\.1:\d+")
 SERVER_LABEL = re.compile(r"[a-z][a-z0-9-]{0,31}")
 # Transport errors name the HTTP stack that failed; only the category is
@@ -75,11 +76,28 @@ def substitute(text, case):
     return text
 
 
-def own_version(binary):
-    """The bare version a binary reports, which its messages embed."""
+def own_identity(binary):
+    """Build metadata from this binary, rather than a different release."""
     output = subprocess.run([binary, "--version"], capture_output=True, text=True, timeout=60, check=False).stdout
-    match = re.match(r"dw (\S+) ", output)
-    return match.group(1) if match else None
+    match = re.match(r"dw (\S+) \(commit ([^,]+), built ([^)]+)\)", output)
+    return dict(zip(("version", "commit", "build_date"), match.groups())) if match else {}
+
+
+def normalize_doctor_identity(stdout_json, stdout_text, identity):
+    """Normalize only metadata that agrees with the binary's --version.
+
+    A wrong diagnostic commit or build date remains a comparison failure.
+    """
+    for key, label in (("commit", "Commit"), ("build_date", "Built")):
+        if key not in identity:
+            continue
+        replacement = f"<{key}>"
+        stdout_text = re.sub(r"^  " + label + ": " + re.escape(identity[key]) + r"$",
+                             f"  {label}: {replacement}", stdout_text, flags=re.M)
+        for item in stdout_json or []:
+            if isinstance(item, dict) and isinstance(item.get("dw"), dict) and item["dw"].get(key) == identity[key]:
+                item["dw"][key] = replacement
+    return stdout_json, stdout_text
 
 
 def normalize_text(text, config_dir, version=None):
@@ -93,6 +111,7 @@ def normalize_text(text, config_dir, version=None):
     text = MARKDOWN_TITLE.sub(r"\1\n<underline>", text)
     text = TIME.sub("<time>", text)
     text = TRANSPORT_LINE.sub("<transport error>", text)
+    text = DOCTOR_TRANSPORT_LINE.sub(r"\1<transport error>", text)
     text = LOOPBACK_PORT.sub("127.0.0.1:<port>", text)
     return ULID.sub("<id>", text)
 
@@ -138,7 +157,9 @@ def read_config(config_dir):
         return normalize_text(handle.read(), config_dir)
 
 
-def run_case(binary, case, env_base, version=None):
+def run_case(binary, case, env_base, identity=None):
+    identity = identity or {}
+    version = identity.get("version")
     config_dir = tempfile.mkdtemp(prefix="dw-parity-")
     try:
         env = dict(env_base)
@@ -153,10 +174,13 @@ def run_case(binary, case, env_base, version=None):
         completed = subprocess.run([binary] + argv, env=env, stdin=subprocess.DEVNULL, capture_output=True,
                                    text=True, timeout=60, check=False)
         stdout_json = parse_json_lines(completed.stdout)
+        stdout_text = completed.stdout
+        if argv and argv[0] == "doctor":
+            stdout_json, stdout_text = normalize_doctor_identity(stdout_json, stdout_text, identity)
         return {
             "exit_code": completed.returncode,
             "stdout_json": normalize_value(stdout_json, config_dir, version) if stdout_json is not None else None,
-            "stdout_text": None if stdout_json is not None else normalize_text(completed.stdout, config_dir, version),
+            "stdout_text": None if stdout_json is not None else normalize_text(stdout_text, config_dir, version),
             "stderr_category": stderr_category(completed.stderr),
             "stderr": normalize_text(completed.stderr, config_dir, version),
             "config": read_config(config_dir),
@@ -300,7 +324,7 @@ def run_all(candidate_binary, servers, verbose):
 
     reference_binary = invocation_path(REFERENCE, "reference")
     candidate_binary = invocation_path(candidate_binary, "candidate")
-    versions = {reference_binary: own_version(reference_binary), candidate_binary: own_version(candidate_binary)}
+    identities = {reference_binary: own_identity(reference_binary), candidate_binary: own_identity(candidate_binary)}
     results = []
     for case in expand_cases(cases, servers):
         uses_mock = "mock" in case or "mock_routes" in case
@@ -308,8 +332,8 @@ def run_all(candidate_binary, servers, verbose):
         if mock:
             case = dict(case, vars={**case.get("vars", {}), "mock": mock.url})
         try:
-            reference = run_case(reference_binary, case, env_base, versions[reference_binary])
-            candidate = run_case(candidate_binary, case, env_base, versions[candidate_binary])
+            reference = run_case(reference_binary, case, env_base, identities[reference_binary])
+            candidate = run_case(candidate_binary, case, env_base, identities[candidate_binary])
         finally:
             if mock:
                 mock.close()
diff --git a/rust/src/build_info.rs b/rust/src/build_info.rs
index f550369..cfa381f 100644
--- a/rust/src/build_info.rs
+++ b/rust/src/build_info.rs
@@ -11,9 +11,20 @@ pub fn version() -> &'static str {
 
 /// `dw <version> (commit <sha>, built <date>)`, as `dw --version` prints it.
 pub fn version_line() -> String {
-    let commit = option_env!("DW_CLI_COMMIT")
+    format!(
+        "dw {} (commit {}, built {})",
+        version(),
+        commit(),
+        build_date()
+    )
+}
+
+pub fn commit() -> &'static str {
+    option_env!("DW_CLI_COMMIT")
         .map(|commit| &commit[..commit.len().min(12)])
-        .unwrap_or("unknown");
-    let built = option_env!("DW_CLI_BUILD_DATE").unwrap_or("source checkout");
-    format!("dw {} (commit {commit}, built {built})", version())
+        .unwrap_or("unknown")
+}
+
+pub fn build_date() -> &'static str {
+    option_env!("DW_CLI_BUILD_DATE").unwrap_or("source checkout")
 }
diff --git a/rust/src/commands/doctor.rs b/rust/src/commands/doctor.rs
new file mode 100644
index 0000000..37df62c
--- /dev/null
+++ b/rust/src/commands/doctor.rs
@@ -0,0 +1,423 @@
+//! Connection and compatibility diagnostics, without the session preflight:
+//! `doctor` must still explain a Server this CLI cannot operate against.
+
+use crate::cli::Input;
+use crate::config::ProfileStore;
+use crate::error::{exit, CliError};
+use crate::formatter::writeln_out;
+use crate::{build_info, client, compat, formatter, output, php};
+use serde_json::{json, Value};
+use std::fmt::Write as _;
+
+pub fn run(input: &Input) -> Result<i32, CliError> {
+    let context = client::Context::new(input)?;
+    let config_path = ProfileStore::new()?.path_string();
+    let effective = context.connection.effective_config(&config_path);
+    let mut diagnostic = json!({
+        "dw": {"version": build_info::version(), "commit": build_info::commit(),
+            "build_date": build_info::build_date()},
+        "connection": {
+            "server": context.connection.server,
+            "namespace": context.connection.namespace,
+            "profile": {"name": effective["profile"]["name"],
+                "source": effective["profile"]["source"], "config_path": config_path},
+            "token": {"present": effective["auth"]["present"], "source": effective["auth"]["source"],
+                "env": effective["auth"]["env"], "profile": effective["auth"]["profile"]},
+            "tls": {"verify": context.connection.tls_verify, "source": effective["tls"]["source"]},
+            "effective_config": effective,
+        },
+        "server": {"reachable": false}, "warnings": [], "recommendations": [],
+    });
+    let code = match context.cluster_info(None) {
+        Ok(info) => {
+            diagnostic["server"] = server_diagnostic(&info);
+            diagnostic["warnings"] = json!(compat::warnings(&info, build_info::version(), true));
+            exit::SUCCESS
+        }
+        Err(error) => {
+            diagnostic["server"]["error"] = json!(error.message());
+            // Local client-construction failures are Throwable in PHP;
+            // transport and HTTP errors retain their ServerException category.
+            match error {
+                CliError::Invalid(_) | CliError::Compatibility(_) => exit::FAILURE,
+                _ => error.exit_code(),
+            }
+        }
+    };
+    diagnostic["recommendations"] = json!(recommendations(&diagnostic, code));
+    if context.mode.machine_readable() {
+        output::print_json(&diagnostic);
+    } else {
+        writeln_out(render(&diagnostic).trim_end_matches('\n'));
+    }
+    Ok(code)
+}
+
+fn scalar(value: &Value, key: &str) -> Value {
+    php::get(value, key)
+        .filter(|value| php::is_scalar(Some(value)))
+        .map(|value| json!(php::to_string(Some(value))))
+        .unwrap_or(Value::Null)
+}
+
+fn integer(value: &Value, key: &str) -> Value {
+    php::get(value, key)
+        .filter(|value| php::is_scalar(Some(value)))
+        .map(|value| json!(php::int_cast(Some(value))))
+        .unwrap_or(Value::Null)
+}
+
+fn array(value: &Value, key: &str) -> Value {
+    php::get(value, key)
+        .filter(|value| php::is_array(value))
+        .cloned()
+        .unwrap_or(Value::Null)
+}
+
+fn auth_diagnostic(info: &Value) -> Value {
+    let contract = array(info, "auth_composition_contract");
+    if contract.is_null() {
+        return Value::Null;
+    }
+    json!({"schema": scalar(&contract, "schema"), "version": integer(&contract, "version"),
+        "scope": scalar(&contract, "scope"),
+        "canonical_environment": array(&contract, "canonical_environment"),
+        "auth_material": array(&contract, "auth_material")})
+}
+
+fn control_diagnostic(info: &Value) -> Value {
+    let control = array(info, "control_plane");
+    if control.is_null() {
+        return Value::Null;
+    }
+    let contract = array(&control, "request_contract");
+    json!({"version": scalar(&control, "version"), "header": scalar(&control, "header"),
+    "request_contract": if contract.is_null() {Value::Null} else {
+        json!({"schema": scalar(&contract, "schema"), "version": scalar(&contract, "version")})
+    }})
+}
+
+fn invocable_diagnostic(worker: &Value) -> Value {
+    let contract = array(worker, "invocable_carrier_contract");
+    if contract.is_null() {
+        return Value::Null;
+    }
+    let kinds = &contract["scope"]["task_kinds"];
+    json!({
+        "schema": scalar(&contract, "schema"), "version": integer(&contract, "version"),
+        "carrier_type": scalar(&contract, "carrier_type"),
+        "task_kinds": if php::is_array(kinds) {json!(php::values(kinds))} else {json!([])},
+        "request_content_type": scalar(&contract["request"], "content_type"),
+        "response_content_type": scalar(&contract["response"], "content_type"),
+        "idempotency_key_source": scalar(&contract["request"], "idempotency_key_source"),
+        "retry_authority": scalar(&contract["rollout_safety"], "retry_authority"),
+    })
+}
+
+fn worker_diagnostic(info: &Value) -> Value {
+    let worker = array(info, "worker_protocol");
+    if worker.is_null() {
+        return Value::Null;
+    }
+    json!({"version": scalar(&worker, "version"),
+        "server_capabilities": worker.get("server_capabilities"),
+        "invocable_carrier_contract": invocable_diagnostic(&worker),
+        "external_task_input_contract": compat::input_diagnostic(info),
+        "external_task_result_contract": compat::result_diagnostic(info)})
+}
+
+fn server_diagnostic(info: &Value) -> Value {
+    json!({"reachable": true, "version": scalar(info, "version"),
+        "server_id": scalar(info, "server_id"), "default_namespace": scalar(info, "default_namespace"),
+        "auth_composition_contract": auth_diagnostic(info),
+        "control_plane": control_diagnostic(info), "worker_protocol": worker_diagnostic(info),
+        "client_compatibility": info.get("client_compatibility"), "cluster_info": info})
+}
+
+fn advice(id: &str, severity: &str, message: &str, command: Option<&str>) -> Value {
+    let mut value = json!({"id": id, "severity": severity, "message": message});
+    if let Some(command) = command {
+        value["command"] = json!(command);
+    }
+    value
+}
+
+fn recommendations(diagnostic: &Value, code: i32) -> Vec<Value> {
+    let mut result = Vec::new();
+    let connection = &diagnostic["connection"];
+    let reachable = diagnostic["server"]["reachable"] == true;
+    if !reachable {
+        result.push(advice(
+            if code == exit::NETWORK {"server.unreachable"} else {"server.error"}, "error",
+            if code == exit::NETWORK {
+                "Check the server URL, DNS, port, and TLS settings for the selected environment."
+            } else {
+                "Inspect the server error, then retry with --output=json if support needs the full diagnostic payload."
+            },
+            Some(&format!("dw doctor --server={} --output=json", text(&connection["server"], "http://localhost:8080"))),
+        ));
+    }
+    if connection["token"]["present"] != true {
+        result.push(advice(
+            "auth.token_missing",
+            "warning",
+            "Set an auth token if the target server requires authenticated operator requests.",
+            Some("dw env:set <name> --token-env=DURABLE_WORKFLOW_AUTH_TOKEN"),
+        ));
+    }
+    if connection["tls"]["verify"] == false {
+        result.push(advice("tls.verification_disabled", "warning",
+            "Re-enable TLS verification before using this profile outside a local or disposable test server.",
+            Some("dw env:set <name> --tls-verify=true")));
+    }
+    if diagnostic["warnings"]
+        .as_array()
+        .is_some_and(|warnings| !warnings.is_empty())
+    {
+        result.push(advice(
+            "compatibility.warning",
+            "warning",
+            "Review protocol compatibility before running mutating workflow or worker commands.",
+            Some("dw server:info --output=json"),
+        ));
+    }
+    if result.is_empty() {
+        result.push(advice("doctor.clean", "info",
+            "Connection, auth discovery, TLS settings, and advertised protocol metadata look ready for CLI operations.", None));
+    }
+    if reachable {
+        result.push(advice("semantics.execution_contract", "info", concat!(
+            "Duplicate activity attempts and replayed workflow tasks are normal v2 behaviour. ",
+            "Consult durable-workflow/workflow:docs/architecture/execution-guarantees.md for the at-least-once, ",
+            "deterministic-replay, and exactly-once-at-the-durable-state-layer guarantees, and ",
+            "use activity_execution_id / workflow_command_id as the framework idempotency surfaces."), None));
+        result.push(advice(
+            "semantics.worker_compatibility_contract",
+            "info",
+            concat!(
+            "Mixed-version deployments are driven by opaque compatibility markers. ",
+            "Consult durable-workflow/workflow:docs/architecture/worker-compatibility.md for the ",
+            "run-pinned-at-start, inheritance-through-retry-continue-as-new-and-child-runs, ",
+            "claim-time-enforcement, and no-compatible-worker-is-an-explicit-operational-state ",
+            "guarantees, plus the operator-driven rollout-and-rollback posture and the ",
+            "heartbeat-TTL ceiling that bound fleet staleness."),
+            None,
+        ));
+    }
+    result
+}
+
+fn text(value: &Value, fallback: &str) -> String {
+    php::or(Some(value), fallback)
+}
+
+/// Only trusted render templates may create terminal formatting or layout.
+/// Keep remote/control text visible as data without changing machine JSON.
+fn display(value: &Value, fallback: &str) -> String {
+    let mut safe = String::new();
+    for character in text(value, fallback).chars() {
+        match character {
+            '\n' => safe.push_str("\\n"),
+            '\r' => safe.push_str("\\r"),
+            '\t' => safe.push_str("\\t"),
+            character if character.is_control() || matches!(character, '\u{2028}' | '\u{2029}') => {
+                write!(safe, "\\u{{{:04x}}}", u32::from(character)).expect("write to string");
+            }
+            character => safe.push(character),
+        }
+    }
+    formatter::escape(&safe)
+}
+
+fn line(out: &mut String, label: &str, value: &Value) {
+    writeln!(out, "  {label}: {}", display(value, "unknown")).expect("write to string");
+}
+
+fn render(diagnostic: &Value) -> String {
+    let mut out = String::from("<info>dw doctor</info>\n");
+    let dw = &diagnostic["dw"];
+    for (label, key) in [
+        ("CLI Version", "version"),
+        ("Commit", "commit"),
+        ("Built", "build_date"),
+    ] {
+        line(&mut out, label, &dw[key]);
+    }
+    out.push_str("\nConnection:\n");
+    let connection = &diagnostic["connection"];
+    line(&mut out, "Server", &connection["server"]);
+    line(&mut out, "Namespace", &connection["namespace"]);
+    let profile = &connection["profile"];
+    let name = if profile["name"].is_null() {
+        "none".into()
+    } else {
+        format!(
+            "{} ({})",
+            text(&profile["name"], "unknown"),
+            text(&profile["source"], "unknown")
+        )
+    };
+    line(&mut out, "Profile", &json!(name));
+    writeln!(
+        out,
+        "  Token: {} via {}",
+        if connection["token"]["present"] == true {
+            "present"
+        } else {
+            "not set"
+        },
+        display(&connection["token"]["source"], "none")
+    )
+    .expect("write to string");
+    writeln!(
+        out,
+        "  TLS Verify: {}",
+        if connection["tls"]["verify"] == false {
+            "no"
+        } else {
+            "yes"
+        }
+    )
+    .expect("write to string");
+    out.push_str("\nServer:\n");
+    render_server(&mut out, &diagnostic["server"]);
+    if let Some(warnings) = diagnostic["warnings"]
+        .as_array()
+        .filter(|warnings| !warnings.is_empty())
+    {
+        out.push_str("\nWarnings:\n");
+        for warning in warnings {
+            writeln!(out, "  {}", display(warning, "")).expect("write to string");
+        }
+    }
+    if let Some(recommendations) = diagnostic["recommendations"]
+        .as_array()
+        .filter(|items| !items.is_empty())
+    {
+        out.push_str("\nNext steps:\n");
+        for recommendation in recommendations {
+            writeln!(out, "  - {}", display(&recommendation["message"], ""))
+                .expect("write to string");
+            if let Some(command) = recommendation.get("command") {
+                writeln!(out, "    Try: {}", display(command, "")).expect("write to string");
+            }
+        }
+    }
+    out
+}
+
+fn render_server(out: &mut String, server: &Value) {
+    if server["reachable"] != true {
+        out.push_str("  Reachable: no\n");
+        line(out, "Error", &server["error"]);
+        return;
+    }
+    out.push_str("  Reachable: yes\n");
+    for (label, key) in [
+        ("Server ID", "server_id"),
+        ("Version", "version"),
+        ("Default Namespace", "default_namespace"),
+    ] {
+        line(out, label, &server[key]);
+    }
+    if !server["control_plane"].is_null() {
+        line(out, "Control Plane", &server["control_plane"]["version"]);
+    }
+    let worker = &server["worker_protocol"];
+    if !worker.is_null() {
+        line(out, "Worker Protocol", &worker["version"]);
+        let invocable = &worker["invocable_carrier_contract"];
+        if !invocable.is_null() {
+            writeln!(
+                out,
+                "  Invocable Carrier: {} v{} ({})",
+                display(&invocable["schema"], "unknown"),
+                display(&invocable["version"], "unknown"),
+                display(&invocable["carrier_type"], "unknown")
+            )
+            .expect("write to string");
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn human_connection_fields_cannot_create_markup_or_layout() {
+        let value = "<error>untrusted</error>\nForged:\u{1b}]0;title\u{7}\u{9b}31m";
+        let diagnostic = json!({
+            "dw": {},
+            "connection": {"server": value, "namespace": value,
+                "profile": {"name": value, "source": value},
+                "token": {"source": value}},
+            "server": {"reachable": false, "error": value},
+            "warnings": [value],
+            "recommendations": [{"message": value, "command": value}]
+        });
+        let human = formatter::format(&render(&diagnostic), false);
+        assert!(human.contains("<error>untrusted</error>\\nForged:"));
+        assert!(!human.contains("\nForged:"));
+        assert!(!human.chars().any(|c| c.is_control() && c != '\n'));
+        assert_eq!(diagnostic["connection"]["namespace"], value);
+    }
+
+    #[test]
+    fn published_metadata_keeps_fixture_summaries_and_order() {
+        let info: Value = serde_json::from_str(include_str!(
+            "../compat/fixtures/cluster-info-php-2.11.8.json"
+        ))
+        .unwrap();
+        let server = server_diagnostic(&info);
+        assert_eq!(server["cluster_info"], info);
+        let worker = &server["worker_protocol"];
+        assert_eq!(
+            worker["external_task_input_contract"]["fixtures"],
+            json!(["workflow_task", "activity_task"])
+        );
+        assert_eq!(worker["external_task_input_contract"]["valid"], true);
+        assert_eq!(worker["external_task_result_contract"]["valid"], true);
+        assert_eq!(
+            worker["invocable_carrier_contract"]["task_kinds"],
+            json!(["activity_task"])
+        );
+    }
+
+    #[test]
+    fn malformed_metadata_is_reported_without_refusing_diagnostics() {
+        let info = json!({"version": false, "server_id": {}, "control_plane": [],
+            "worker_protocol": {"version": 7, "external_task_input_contract": {"version": "2x", "fixtures": {"unknown": {}, "activity_task": {}}}}});
+        let server = server_diagnostic(&info);
+        assert_eq!(server["reachable"], true);
+        assert_eq!(server["version"], "");
+        assert_eq!(server["server_id"], Value::Null);
+        assert_eq!(server["control_plane"]["version"], Value::Null);
+        let input = &server["worker_protocol"]["external_task_input_contract"];
+        assert_eq!(input["version"], 2);
+        assert_eq!(input["fixtures"], json!(["activity_task"]));
+        assert_eq!(input["valid"], false);
+    }
+
+    #[test]
+    fn failed_connection_keeps_action_order_and_exit_category() {
+        let diagnostic = json!({"connection": {"server": "http://127.0.0.1:9", "token": {"present": false}, "tls": {"verify": false}},
+            "server": {"reachable": false}, "warnings": []});
+        let advice = recommendations(&diagnostic, exit::NETWORK);
+        assert_eq!(
+            advice
+                .iter()
+                .map(|item| item["id"].as_str().unwrap())
+                .collect::<Vec<_>>(),
+            vec![
+                "server.unreachable",
+                "auth.token_missing",
+                "tls.verification_disabled"
+            ]
+        );
+        assert_eq!(
+            recommendations(&diagnostic, exit::AUTH)[0]["id"],
+            "server.error"
+        );
+    }
+}
diff --git a/rust/src/commands/mod.rs b/rust/src/commands/mod.rs
index 4963d07..c0402de 100644
--- a/rust/src/commands/mod.rs
+++ b/rust/src/commands/mod.rs
@@ -1,5 +1,6 @@
 pub mod completion;
 mod completion_log;
+pub mod doctor;
 pub mod env;
 pub mod info;
 pub mod schema;
diff --git a/rust/src/compat.rs b/rust/src/compat.rs
index 769901a..64bd61a 100644
--- a/rust/src/compat.rs
+++ b/rust/src/compat.rs
@@ -13,6 +13,7 @@ mod input_envelope;
 mod request_contract;
 pub mod version;
 
+pub use external_task::{input_diagnostic, result_diagnostic};
 pub use request_contract::RequestContract;
 
 use crate::php;
diff --git a/rust/src/compat/external_task.rs b/rust/src/compat/external_task.rs
index a32bc11..4c9a77a 100644
--- a/rust/src/compat/external_task.rs
+++ b/rust/src/compat/external_task.rs
@@ -4,6 +4,7 @@
 
 use super::{input_envelope, manifest_check, sha256_hex};
 use crate::php;
+use serde_json::json;
 use serde_json::Value;
 
 const VERSION: i64 = 1;
@@ -111,6 +112,56 @@ pub fn result_warning(cluster_info: &Value) -> Option<String> {
     contract_warning(cluster_info, &RESULT)
 }
 
+/// The same fixture summary `doctor` exposes in 2.x, retaining the
+/// Server's fixture order and validating the original manifest.
+fn diagnostic(cluster_info: &Value, contract: &Contract, names: &[(&str, &str)]) -> Value {
+    let Some(manifest) = php::array_get(php::get(cluster_info, "worker_protocol"), contract.key)
+        .filter(|value| php::is_array(value))
+    else {
+        return Value::Null;
+    };
+    let fixtures: Vec<_> = php::get(manifest, "fixtures")
+        .filter(|value| php::is_array(value))
+        .map(php::entries)
+        .unwrap_or_default()
+        .into_iter()
+        .map(|(key, _)| key.text())
+        .filter(|key| names.iter().any(|(name, _)| *name == key))
+        .collect();
+    json!({
+        "schema": php::get(manifest, "schema").filter(|value| php::is_scalar(Some(value)))
+            .map(|value| php::to_string(Some(value))),
+        "version": php::get(manifest, "version").filter(|value| php::is_scalar(Some(value)))
+            .map(|value| php::int_cast(Some(value))),
+        "fixtures": fixtures,
+        "valid": (contract.validate)(manifest).is_empty(),
+    })
+}
+
+pub fn input_diagnostic(cluster_info: &Value) -> Value {
+    let mut summary = diagnostic(cluster_info, &INPUT, &INPUT_FIXTURES);
+    if !summary.is_null() {
+        let scope = &cluster_info["worker_protocol"][INPUT.key]["scope"];
+        // Insert the scope fields before valid, matching the PHP summary.
+        let valid = summary
+            .as_object_mut()
+            .expect("diagnostic object")
+            .remove("valid");
+        summary["activity_grade_fixture_keys"] = json!(scope_fixture_keys(
+            scope,
+            "activity_grade_external_execution"
+        ));
+        summary["worker_protocol_fixture_keys"] =
+            json!(scope_fixture_keys(scope, "worker_protocol_runtime"));
+        summary["valid"] = valid.unwrap_or(Value::Bool(false));
+    }
+    summary
+}
+
+pub fn result_diagnostic(cluster_info: &Value) -> Value {
+    diagnostic(cluster_info, &RESULT, &RESULT_FIXTURES)
+}
+
 fn contract_warning(cluster_info: &Value, contract: &Contract) -> Option<String> {
     if !server_requires(cluster_info, contract.key) && !has_published(cluster_info, contract.key) {
         return None;
diff --git a/rust/src/config.rs b/rust/src/config.rs
index 61b353a..041a53b 100644
--- a/rust/src/config.rs
+++ b/rust/src/config.rs
@@ -371,6 +371,59 @@ pub struct Connection {
     pub token: Option<String>,
     pub tls_verify: bool,
     pub profile: Option<Profile>,
+    sources: ConnectionSources,
+}
+
+#[derive(Debug)]
+struct ConnectionSources {
+    server: &'static str,
+    namespace: &'static str,
+    profile: Option<&'static str>,
+    token: &'static str,
+    token_env: Option<String>,
+    tls: &'static str,
+}
+
+impl Connection {
+    /// Redacted provenance captured while resolving the connection. Never
+    /// include the bearer token in an operator diagnostic.
+    pub fn effective_config(&self, config_path: &str) -> Value {
+        let sources = &self.sources;
+        let profile_name = self.profile.as_ref().map(|profile| &profile.name);
+        let server = json!({"value": self.server, "source": sources.server,
+            "contract_source": contract_source(sources.server)});
+        json!({
+            "contract": {"schema": "durable-workflow.v2.auth-composition.contract", "version": 1},
+            "server": server, "server_url": server,
+            "namespace": {"value": self.namespace, "source": sources.namespace,
+                "contract_source": contract_source(sources.namespace)},
+            "profile": {"name": profile_name, "source": sources.profile,
+                "contract_source": sources.profile.map(|source| match source {
+                    "flag" => "flag_env", "current_env" => "current_profile", other => other,
+                }), "config_path": config_path},
+            "auth": {"type": if self.token.is_some() {"token"} else {"none"},
+                "transport": self.token.as_ref().map(|_| "bearer_authorization_header"),
+                "present": self.token.is_some(), "source": sources.token,
+                "contract_source": match sources.token {
+                    "DURABLE_WORKFLOW_AUTH_TOKEN" => "environment",
+                    "profile_literal" => "selected_profile", other => other,
+                },
+                "env": sources.token_env,
+                "profile": if sources.token.starts_with("profile_") {profile_name} else {None},
+                "value": self.token.as_ref().map(|_| "redacted")},
+            "tls": {"verify": self.tls_verify, "source": sources.tls,
+                "contract_source": contract_source(sources.tls)},
+            "identity": {"subject": null, "roles": [], "source": "server", "asserted": false},
+        })
+    }
+}
+
+fn contract_source(source: &str) -> &str {
+    match source {
+        "profile" => "selected_profile",
+        name if name.starts_with("DURABLE_WORKFLOW_") => "environment",
+        other => other,
+    }
 }
 
 /// Flag values that were written with a non-empty value. An empty or
@@ -383,46 +436,104 @@ pub struct ConnectionFlags<'a> {
     pub tls_verify: Option<&'a str>,
 }
 
+/// Runtime URLs are connection locations, not an implicit credential source.
+/// Refuse before client construction and never echo the rejected value.
+fn validate_runtime_url(server: &str) -> Result<(), CliError> {
+    let invalid = || {
+        CliError::Invalid(
+        "Invalid server URL: use an absolute HTTP(S) runtime base URL without user information, a query, a fragment, or whitespace. Supply authentication through a token option or configured token source."
+            .into(),
+    )
+    };
+    if server
+        .chars()
+        .any(|character| character.is_control() || character.is_whitespace())
+    {
+        return Err(invalid());
+    }
+    let url = reqwest::Url::parse(server).map_err(|_| invalid())?;
+    if !matches!(url.scheme(), "http" | "https")
+        || url.host_str().is_none()
+        || !url.username().is_empty()
+        || url.password().is_some()
+        || url.query().is_some()
+        || url.fragment().is_some()
+    {
+        return Err(invalid());
+    }
+    Ok(())
+}
+
 pub fn resolve(store: &ProfileStore, flags: &ConnectionFlags) -> Result<Connection, CliError> {
-    let profile = if let Some(env) = flags.env {
-        Some(store.require(env, "--env")?)
+    let (profile, profile_source) = if let Some(env) = flags.env {
+        (Some(store.require(env, "--env")?), Some("flag"))
     } else if let Some(env) = env_string("DW_ENV") {
-        Some(store.require(&env, "DW_ENV")?)
+        (Some(store.require(&env, "DW_ENV")?), Some("DW_ENV"))
     } else if let Some(current) = store.load()?.current_env {
-        Some(store.require(&current, "dw env:use (current_env in config)")?)
+        (
+            Some(store.require(&current, "dw env:use (current_env in config)")?),
+            Some("current_env"),
+        )
     } else {
-        None
+        (None, None)
     };
 
-    let pick = |flag: Option<&str>, env: &str, profile_value: Option<&String>, default: &str| {
-        flag.map(str::to_string)
-            .or_else(|| env_string(env))
-            .or_else(|| profile_value.cloned())
-            .unwrap_or_else(|| default.to_string())
-    };
+    let pick =
+        |flag: Option<&str>, env: &'static str, profile_value: Option<&String>, default: &str| {
+            if let Some(value) = flag {
+                (value.to_string(), "flag")
+            } else if let Some(value) = env_string(env) {
+                (value, env)
+            } else if let Some(value) = profile_value {
+                (value.clone(), "profile")
+            } else {
+                (default.to_string(), "default")
+            }
+        };
 
-    let server = pick(
+    let (server, server_source) = pick(
         flags.server,
         "DURABLE_WORKFLOW_SERVER_URL",
         profile.as_ref().and_then(|p| p.server.as_ref()),
         "http://localhost:8080",
     );
-    let namespace = pick(
+    validate_runtime_url(&server)?;
+    let (namespace, namespace_source) = pick(
         flags.namespace,
         "DURABLE_WORKFLOW_NAMESPACE",
         profile.as_ref().and_then(|p| p.namespace.as_ref()),
         "default",
     );
-    let token = flags
-        .token
-        .map(str::to_string)
-        .or_else(|| env_string("DURABLE_WORKFLOW_AUTH_TOKEN"))
-        .or_else(|| profile.as_ref().and_then(Profile::resolve_token));
-    let tls_verify = match flags.tls_verify {
-        Some(value) => parse_boolean(value, "--tls-verify")?,
+    let (token, token_source, token_env) = if let Some(value) = flags.token {
+        (Some(value.to_string()), "flag", None)
+    } else if let Some(value) = env_string("DURABLE_WORKFLOW_AUTH_TOKEN") {
+        (
+            Some(value),
+            "DURABLE_WORKFLOW_AUTH_TOKEN",
+            Some("DURABLE_WORKFLOW_AUTH_TOKEN".into()),
+        )
+    } else {
+        let resolved = profile.as_ref().and_then(Profile::resolve_token);
+        match profile
+            .as_ref()
+            .and_then(|profile| profile.token_source.as_ref())
+        {
+            Some(TokenSource::Env(name)) => (resolved, "profile_env", Some(name.clone())),
+            Some(TokenSource::Literal(_)) => (resolved, "profile_literal", None),
+            None => (None, "none", None),
+        }
+    };
+    let (tls_verify, tls_source) = match flags.tls_verify {
+        Some(value) => (parse_boolean(value, "--tls-verify")?, "flag"),
         None => match env_string("DURABLE_WORKFLOW_TLS_VERIFY") {
-            Some(value) => parse_boolean(&value, "DURABLE_WORKFLOW_TLS_VERIFY")?,
-            None => profile.as_ref().map(|p| p.tls_verify).unwrap_or(true),
+            Some(value) => (
+                parse_boolean(&value, "DURABLE_WORKFLOW_TLS_VERIFY")?,
+                "DURABLE_WORKFLOW_TLS_VERIFY",
+            ),
+            None => profile
+                .as_ref()
+                .map(|p| (p.tls_verify, "profile"))
+                .unwrap_or((true, "default")),
         },
     };
 
@@ -432,6 +543,14 @@ pub fn resolve(store: &ProfileStore, flags: &ConnectionFlags) -> Result<Connecti
         token,
         tls_verify,
         profile,
+        sources: ConnectionSources {
+            server: server_source,
+            namespace: namespace_source,
+            profile: profile_source,
+            token: token_source,
+            token_env,
+            tls: tls_source,
+        },
     })
 }
 
@@ -449,6 +568,35 @@ pub fn parse_boolean(value: &str, source: &str) -> Result<bool, CliError> {
 mod tests {
     use super::*;
 
+    #[test]
+    fn runtime_urls_keep_encoded_paths_but_refuse_implicit_credentials() {
+        for server in [
+            "http://127.0.0.1:8080",
+            "https://example.test/api/runtime/v1/namespaces/tenant-1",
+            "http://[::1]:8080/runtime/%3Fname%23value",
+        ] {
+            assert!(validate_runtime_url(server).is_ok());
+        }
+        for server in [
+            "https://fixture-user:fixture-password@example.test",
+            "https://fixture-user@example.test",
+            "https://:fixture-password@example.test",
+            "https://example.test?token=fixture-query",
+            "https://example.test#fixture-fragment",
+            "https://example.test?",
+            "https://example.test#",
+            "https://example.test/has space",
+            "https://example.test/\nvalue",
+            "https://example.test/\u{9b}value",
+            "relative/runtime",
+            "ftp://example.test",
+        ] {
+            let error = validate_runtime_url(server).unwrap_err();
+            assert_eq!(error.exit_code(), crate::error::exit::INVALID);
+            assert!(!error.message().contains("fixture-"));
+        }
+    }
+
     #[test]
     fn describe_redacts_literal_tokens() {
         let profile = Profile {
@@ -492,4 +640,44 @@ mod tests {
         assert!(!parse_boolean("off", "x").unwrap());
         assert!(parse_boolean("maybe", "x").is_err());
     }
+
+    #[test]
+    fn effective_config_redacts_auth_and_preserves_source_contracts() {
+        let connection = Connection {
+            server: "https://example.test".into(),
+            namespace: "tenant".into(),
+            token: Some("synthetic-private-value".into()),
+            tls_verify: false,
+            profile: Some(Profile {
+                name: "prod".into(),
+                server: None,
+                namespace: None,
+                token_source: None,
+                tls_verify: false,
+                output: None,
+            }),
+            sources: ConnectionSources {
+                server: "profile",
+                namespace: "DURABLE_WORKFLOW_NAMESPACE",
+                profile: Some("current_env"),
+                token: "profile_env",
+                token_env: Some("TEST_TOKEN".into()),
+                tls: "flag",
+            },
+        };
+        let effective = connection.effective_config("config.json");
+        assert!(!effective.to_string().contains("synthetic-private-value"));
+        assert_eq!(effective["server"], effective["server_url"]);
+        assert_eq!(effective["server"]["contract_source"], "selected_profile");
+        assert_eq!(effective["namespace"]["contract_source"], "environment");
+        assert_eq!(effective["profile"]["contract_source"], "current_profile");
+        assert_eq!(
+            effective["auth"],
+            json!({"type": "token", "transport": "bearer_authorization_header",
+            "present": true, "source": "profile_env", "contract_source": "profile_env",
+            "env": "TEST_TOKEN", "profile": "prod", "value": "redacted"})
+        );
+        assert_eq!(effective["tls"]["source"], "flag");
+        assert_eq!(effective["identity"]["asserted"], false);
+    }
 }
diff --git a/rust/src/main.rs b/rust/src/main.rs
index f5e1ead..4a9a9e0 100644
--- a/rust/src/main.rs
+++ b/rust/src/main.rs
@@ -32,10 +32,11 @@ enum Handler {
 /// Commands ported so far. Every other catalog command is recognized and
 /// reported as not yet available.
 fn handler(name: &str) -> Option<Handler> {
-    use commands::{completion, env, schema, server, upgrade};
+    use commands::{completion, doctor, env, schema, server, upgrade};
     Some(match name {
         "completion" => Handler::Local(completion::completion),
         "_complete" => Handler::Local(completion::complete),
+        "doctor" => Handler::Local(doctor::run),
         "schema:list" => Handler::Local(schema::list),
         "schema:manifest" => Handler::Local(schema::manifest),
         "schema:show" => Handler::Local(schema::show),
diff --git a/rust/tests/doctor_security.rs b/rust/tests/doctor_security.rs
new file mode 100644
index 0000000..9f922ec
--- /dev/null
+++ b/rust/tests/doctor_security.rs
@@ -0,0 +1,296 @@
+use serde_json::{json, Value};
+use std::fs;
+use std::io::{Read, Write};
+use std::net::TcpListener;
+use std::path::PathBuf;
+use std::process::{Command, Output, Stdio};
+use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering};
+use std::sync::Arc;
+use std::thread::{self, JoinHandle};
+use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
+
+struct Scratch(PathBuf);
+
+impl Scratch {
+    fn new() -> Self {
+        static NEXT: AtomicU64 = AtomicU64::new(0);
+        let now = SystemTime::now()
+            .duration_since(UNIX_EPOCH)
+            .unwrap()
+            .as_nanos();
+        for _ in 0..64 {
+            let count = NEXT.fetch_add(1, Ordering::Relaxed);
+            let path = std::env::temp_dir().join(format!(
+                "dw-doctor-security-{}-{now}-{count}",
+                std::process::id()
+            ));
+            let mut builder = fs::DirBuilder::new();
+            #[cfg(unix)]
+            {
+                use std::os::unix::fs::DirBuilderExt;
+                builder.mode(0o700);
+            }
+            match builder.create(&path) {
+                Ok(()) => return Self(path),
+                Err(error) if error.kind() == std::io::ErrorKind::AlreadyExists => continue,
+                Err(error) => panic!("cannot create doctor test directory: {error}"),
+            }
+        }
+        panic!("doctor test directory collisions exceeded the retry limit");
+    }
+
+    fn command(&self) -> Command {
+        let binary = std::env::var_os("DW_DOCTOR_TEST_BINARY")
+            .map(PathBuf::from)
+            .unwrap_or_else(|| env!("CARGO_BIN_EXE_dw").into());
+        let mut command = Command::new(binary);
+        command
+            .arg("doctor")
+            .current_dir(&self.0)
+            .env_clear()
+            .env("HOME", &self.0)
+            .env("DW_CONFIG_HOME", &self.0)
+            .env("PATH", std::env::var_os("PATH").unwrap_or_default());
+        command
+    }
+
+    fn profile(&self, name: &str, url: &str) {
+        fs::write(
+            self.0.join("config.json"),
+            serde_json::to_vec(&json!({
+                "schema": "durable-workflow.cli.config", "version": 1,
+                "current_env": name, "envs": {(name): {"server": url}}
+            }))
+            .unwrap(),
+        )
+        .unwrap();
+    }
+}
+
+impl Drop for Scratch {
+    fn drop(&mut self) {
+        let _ = fs::remove_dir_all(&self.0);
+    }
+}
+
+struct Server {
+    url: String,
+    requests: Arc<AtomicUsize>,
+    stop: Arc<AtomicBool>,
+    thread: Option<JoinHandle<()>>,
+}
+
+impl Server {
+    fn new(status: u16, body: Value) -> Self {
+        let listener = TcpListener::bind("127.0.0.1:0").unwrap();
+        let url = format!("http://{}", listener.local_addr().unwrap());
+        listener.set_nonblocking(true).unwrap();
+        let requests = Arc::new(AtomicUsize::new(0));
+        let stop = Arc::new(AtomicBool::new(false));
+        let observed = requests.clone();
+        let stopped = stop.clone();
+        let body = serde_json::to_vec(&body).unwrap();
+        let thread = thread::spawn(move || {
+            while !stopped.load(Ordering::Relaxed) {
+                match listener.accept() {
+                    Ok((mut stream, _)) => {
+                        observed.fetch_add(1, Ordering::Relaxed);
+                        stream
+                            .set_read_timeout(Some(Duration::from_secs(1)))
+                            .unwrap();
+                        let mut buffer = [0; 4096];
+                        let _ = stream.read(&mut buffer);
+                        let header = format!(
+                            "HTTP/1.1 {status} Test\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n",
+                            body.len()
+                        );
+                        let _ = stream.write_all(header.as_bytes());
+                        let _ = stream.write_all(&body);
+                    }
+                    Err(error) if error.kind() == std::io::ErrorKind::WouldBlock => {
+                        thread::sleep(Duration::from_millis(2));
+                    }
+                    Err(error) => panic!("doctor fixture accept failed: {error}"),
+                }
+            }
+        });
+        Self {
+            url,
+            requests,
+            stop,
+            thread: Some(thread),
+        }
+    }
+}
+
+impl Drop for Server {
+    fn drop(&mut self) {
+        self.stop.store(true, Ordering::Relaxed);
+        self.thread.take().unwrap().join().unwrap();
+    }
+}
+
+fn bounded(mut command: Command) -> Output {
+    let mut child = command
+        .stdout(Stdio::piped())
+        .stderr(Stdio::piped())
+        .spawn()
+        .unwrap();
+    let deadline = Instant::now() + Duration::from_secs(5);
+    while child.try_wait().unwrap().is_none() {
+        if Instant::now() > deadline {
+            let _ = child.kill();
+            let _ = child.wait();
+            panic!("doctor exceeded the fixture deadline");
+        }
+        thread::sleep(Duration::from_millis(5));
+    }
+    child.wait_with_output().unwrap()
+}
+
+fn credential_cases(source: &str) {
+    let server = Server::new(200, json!({"server_id": "fixture"}));
+    let closed = TcpListener::bind("127.0.0.1:0").unwrap();
+    let unavailable = format!("http://{}", closed.local_addr().unwrap());
+    drop(closed);
+    for base in [&server.url, &unavailable] {
+        let authority = base.strip_prefix("http://").unwrap();
+        for url in [
+            format!("http://fixture-url-user:fixture-url-password@{authority}/runtime"),
+            format!("http://fixture-url-user@{authority}/runtime"),
+            format!("{base}/runtime?token=fixture-url-query"),
+            format!("{base}/runtime?signature=fixture-url-signature"),
+            format!("{base}/runtime#fixture-url-fragment"),
+        ] {
+            for mode in ["--no-ansi", "--output=json", "--output=jsonl"] {
+                let scratch = Scratch::new();
+                let mut command = scratch.command();
+                command.arg(mode);
+                match source {
+                    "flag" => {
+                        command.arg(format!("--server={url}"));
+                    }
+                    "environment" => {
+                        command.env("DURABLE_WORKFLOW_SERVER_URL", &url);
+                    }
+                    "profile" => {
+                        scratch.profile("target", &url);
+                        command.arg("--env=target");
+                    }
+                    _ => unreachable!(),
+                }
+                let output = bounded(command);
+                for bytes in [&output.stdout, &output.stderr] {
+                    let text = String::from_utf8_lossy(bytes);
+                    assert!(
+                        !text.contains("fixture-url-"),
+                        "URL credential leaked via {source} {mode}: {text}"
+                    );
+                }
+                assert_eq!(
+                    output.status.code(),
+                    Some(2),
+                    "credential URL must refuse before diagnostics: {source} {mode}"
+                );
+                if mode != "--no-ansi" {
+                    let data: Value = serde_json::from_slice(&output.stdout).unwrap();
+                    assert!(!data.to_string().contains("fixture-url-"));
+                    assert_eq!(data["exit_code"], 2);
+                }
+            }
+        }
+    }
+    assert_eq!(
+        server.requests.load(Ordering::Relaxed),
+        0,
+        "credential-bearing runtime URL reached the network"
+    );
+}
+
+#[test]
+fn flag_url_credentials_are_refused_without_echo_or_request() {
+    credential_cases("flag");
+}
+
+#[test]
+fn environment_url_credentials_are_refused_without_echo_or_request() {
+    credential_cases("environment");
+}
+
+#[test]
+fn profile_url_credentials_are_refused_without_echo_or_request() {
+    credential_cases("profile");
+}
+
+fn attack() -> &'static str {
+    "<error>UNTRUSTED</error>\nFORGED_HEADING:\r\t\u{1b}[2J\u{1b}]8;;https://fixture.invalid\u{7}link\u{1b}]8;;\u{7}\u{85}\u{9b}31m\u{2028}\u{2029}"
+}
+
+fn assert_safe_human(output: &Output, code: i32) {
+    assert_eq!(output.status.code(), Some(code));
+    let text = String::from_utf8(output.stdout.clone()).unwrap();
+    let plain = text.replace("\u{1b}[32m", "").replace("\u{1b}[39m", "");
+    assert!(
+        plain.contains("<error>UNTRUSTED</error>"),
+        "remote formatter markup was interpreted: {plain:?}"
+    );
+    assert!(
+        !plain.chars().any(|c| c.is_control() && c != '\n'),
+        "remote terminal controls survived: {plain:?}"
+    );
+    assert!(!plain.contains(['\u{2028}', '\u{2029}']));
+    assert!(
+        !plain.contains("\nFORGED_HEADING:"),
+        "remote text forged a diagnostic section"
+    );
+    assert!(output.stderr.is_empty());
+}
+
+#[test]
+fn cluster_info_is_safe_in_human_output_and_exact_in_json() {
+    let text = attack();
+    let info = json!({
+        "server_id": text, "version": text, "default_namespace": text,
+        "control_plane": {"version": text},
+        "worker_protocol": {"version": text, "invocable_carrier_contract": {
+            "schema": text, "version": 1, "carrier_type": text
+        }}
+    });
+    let server = Server::new(200, info.clone());
+    for mode in ["--no-ansi", "--ansi", "--output=json", "--output=jsonl"] {
+        let scratch = Scratch::new();
+        scratch.profile(text, &server.url);
+        let mut command = scratch.command();
+        command.arg(mode);
+        let output = bounded(command);
+        if mode.starts_with("--output=") {
+            assert_eq!(output.status.code(), Some(0));
+            let data: Value = serde_json::from_slice(&output.stdout).unwrap();
+            assert_eq!(data["server"]["cluster_info"], info);
+            assert_eq!(data["connection"]["profile"]["name"], text);
+        } else {
+            assert_safe_human(&output, 0);
+        }
+    }
+}
+
+#[test]
+fn http_error_text_is_safe_in_human_output_and_exact_in_json() {
+    let server = Server::new(500, json!({"message": attack()}));
+    for mode in ["--no-ansi", "--ansi", "--output=json", "--output=jsonl"] {
+        let scratch = Scratch::new();
+        let mut command = scratch.command();
+        command.arg(format!("--server={}", server.url)).arg(mode);
+        let output = bounded(command);
+        if mode.starts_with("--output=") {
+            assert_eq!(output.status.code(), Some(6));
+            let data: Value = serde_json::from_slice(&output.stdout).unwrap();
+            assert_eq!(
+                data["server"]["error"],
+                format!("Server error: {}", attack())
+            );
+        } else {
+            assert_safe_human(&output, 6);
+        }
+    }
+}