fix(config): preserve profile auth when overriding model

/provider and /model are registered with remote_invocable=False and
remote_admin_opt_in=True in src/openharness/commands/registry.py, but
OhmoSessionRuntimePool.stream_message intercepted them with
_handle_gateway_scoped_command before the remote-allowed gate ran, so the
contract was silently skipped for both commands.

Move the gateway-scoped intercept after the gating block so the existing
remote_admin_opt_in / allow_remote_admin_commands +
allowed_remote_admin_commands path governs them like every other admin
command. Add a regression test that asserts /provider and /model are
rejected unless the operator has opted in, and update the existing
positive test to set the opt-in so it continues to exercise the success
path.

Refs #280

Co-authored-by: glitch-ux <glitch-ux@users.noreply.github.com>

ohmo/cli.py

@@ -25,6 +25,7 @@ from ohmo.runtime import launch_ohmo_react_tui, run_ohmo_backend, run_ohmo_print
 from ohmo.session_storage import OhmoSessionBackend
 from ohmo.workspace import (
     get_gateway_config_path,
+    get_logs_dir,
     get_workspace_root,
     get_soul_path,
     get_state_path,
@@ -407,18 +408,42 @@ def _maybe_restart_gateway(*, cwd: str | Path, workspace: str | Path) -> None:
     print(f"ohmo gateway restarted (pid={pid})")
 
 
-def _configure_gateway_logging(workspace: str | Path | None = None) -> None:
+def _configure_gateway_logging(
+    workspace: str | Path | None = None,
+    *,
+    console: bool = True,
+    log_file: bool = True,
+) -> None:
     """Configure foreground gateway logging."""
     config = load_gateway_config(workspace)
     level_name = str(config.log_level or "INFO").upper()
     level = getattr(logging, level_name, logging.INFO)
+    handlers = _build_gateway_logging_handlers(workspace, console=console, log_file=log_file)
     logging.basicConfig(
         level=level,
         format="%(asctime)s [%(name)s] %(levelname)s %(message)s",
+        handlers=handlers,
         force=True,
     )
 
 
+def _build_gateway_logging_handlers(
+    workspace: str | Path | None = None,
+    *,
+    console: bool,
+    log_file: bool,
+) -> list[logging.Handler]:
+    """Build gateway log handlers for foreground and daemon modes."""
+    handlers: list[logging.Handler] = []
+    if console:
+        handlers.append(logging.StreamHandler())
+    if log_file:
+        log_path = get_logs_dir(workspace) / "gateway.log"
+        log_path.parent.mkdir(parents=True, exist_ok=True)
+        handlers.append(logging.FileHandler(log_path, encoding="utf-8", delay=True))
+    return handlers
+
+
 @app.callback(invoke_without_command=True)
 def main(
     ctx: typer.Context,
@@ -637,9 +662,11 @@ def user_edit_cmd(
 def gateway_run_cmd(
     cwd: str = typer.Option(str(Path.cwd()), "--cwd", help="Project working directory"),
     workspace: str | None = typer.Option(None, "--workspace", help=_WORKSPACE_HELP),
+    console_log: bool = typer.Option(True, "--console-log/--no-console-log", hidden=True),
+    log_file: bool = typer.Option(True, "--log-file/--no-log-file", hidden=True),
 ) -> None:
     """Run the ohmo gateway in the foreground."""
-    _configure_gateway_logging(workspace)
+    _configure_gateway_logging(workspace, console=console_log, log_file=log_file)
     service = OhmoGatewayService(cwd, workspace)
     raise SystemExit(asyncio.run(service.run_foreground()))
 

ohmo/gateway/bridge.py

@@ -257,9 +257,13 @@ class OhmoGatewayBridge:
                 inbound_meta["message_id"] = message.metadata["message_id"]
         try:
             reply = ""
+            final_media: list[str] = []
+            final_metadata: dict[str, object] = {}
             async for update in self._runtime_pool.stream_message(message, session_key):
                 if update.kind == "final":
                     reply = update.text
+                    final_media = list(getattr(update, "media", None) or (update.metadata or {}).get("_media") or [])
+                    final_metadata = dict(update.metadata or {})
                     continue
                 if not update.text:
                     continue
@@ -319,7 +323,8 @@ class OhmoGatewayBridge:
                 channel=message.channel,
                 chat_id=message.chat_id,
                 content=reply,
-                metadata={**inbound_meta, "_session_key": session_key},
+                media=final_media,
+                metadata={**inbound_meta, **final_metadata, "_session_key": session_key},
             )
         )
 

ohmo/gateway/runtime.py

@@ -9,6 +9,7 @@ import mimetypes
 from pathlib import Path
 import json
 import os
+import re
 import string
 
 from openharness.channels.bus.events import InboundMessage
@@ -62,6 +63,10 @@ _CHANNEL_THINKING_PHRASES_EN = (
 _TEXT_PREVIEW_BYTES = 4096
 _TEXT_PREVIEW_CHARS = 900
 _BINARY_HEAD_BYTES = 32
+_FINAL_REPLY_IMAGE_PATH_RE = re.compile(
+    r"(?P<path>(?:[A-Za-z]:[\\/]|/)[^\r\n`\"'<>|?*\x00]+?\.(?:png|jpe?g|webp|gif|bmp))",
+    re.IGNORECASE,
+)
 _IMAGE_FALLBACK_NOTE = (
     "[Image attachment omitted because the active model does not support image input. "
     "Use the attachment paths and summaries above if needed.]"
@@ -260,19 +265,6 @@ class OhmoSessionRuntimePool:
         if parsed is not None and not message.media:
             command, args = parsed
             command_name = str(getattr(command, "name", "") or "")
-            gateway_result = self._handle_gateway_scoped_command(command_name, args)
-            if gateway_result is not None:
-                message_text, refresh_runtime = gateway_result
-                result = CommandResult(message=message_text, refresh_runtime=refresh_runtime)
-                async for update in self._stream_command_result(
-                    bundle=bundle,
-                    message=message,
-                    session_key=session_key,
-                    user_prompt=user_prompt,
-                    result=result,
-                ):
-                    yield update
-                return
             remote_allowed = getattr(command, "remote_invocable", True)
             if not remote_allowed and self._remote_admin_allowed(command):
                 remote_allowed = True
@@ -296,6 +288,19 @@ class OhmoSessionRuntimePool:
                 ):
                     yield update
                 return
+            gateway_result = self._handle_gateway_scoped_command(command_name, args)
+            if gateway_result is not None:
+                message_text, refresh_runtime = gateway_result
+                result = CommandResult(message=message_text, refresh_runtime=refresh_runtime)
+                async for update in self._stream_command_result(
+                    bundle=bundle,
+                    message=message,
+                    session_key=session_key,
+                    user_prompt=user_prompt,
+                    result=result,
+                ):
+                    yield update
+                return
             result = await command.handler(
                 args,
                 get_command_context(),
@@ -405,6 +410,7 @@ class OhmoSessionRuntimePool:
     ):
         bundle.engine.set_system_prompt(self._runtime_system_prompt(bundle, user_prompt))
         reply_parts: list[str] = []
+        emitted_media: set[str] = set()
         yield GatewayStreamUpdate(
             kind="progress",
             text=_format_channel_progress(
@@ -450,6 +456,7 @@ class OhmoSessionRuntimePool:
                             content=user_prompt,
                             reply_parts=reply_parts,
                         ):
+                            _remember_update_media(emitted_media, update)
                             yield update
                     break
                 async for update in self._convert_stream_event(
@@ -460,6 +467,7 @@ class OhmoSessionRuntimePool:
                     content=user_prompt,
                     reply_parts=reply_parts,
                 ):
+                    _remember_update_media(emitted_media, update)
                     yield update
         except MaxTurnsExceeded as exc:
             yield GatewayStreamUpdate(
@@ -483,10 +491,15 @@ class OhmoSessionRuntimePool:
                 bundle.session_id,
                 _content_snippet(reply),
             )
+            final_media = _extract_final_reply_media(reply, emitted_media)
+            metadata: dict[str, object] = {"_session_key": session_key}
+            if final_media:
+                metadata.update({"_media": final_media, "_final_media_fallback": True})
             yield GatewayStreamUpdate(
                 kind="final",
                 text=reply,
-                metadata={"_session_key": session_key},
+                metadata=metadata,
+                media=final_media or None,
             )
 
     async def _convert_stream_event(
@@ -820,6 +833,46 @@ def _extract_tool_media(event: ToolExecutionCompleted) -> list[str]:
     return media
 
 
+def _remember_update_media(seen: set[str], update: GatewayStreamUpdate) -> None:
+    """Track media already emitted during this gateway turn."""
+    raw_media = update.media or (update.metadata or {}).get("_media") or []
+    if isinstance(raw_media, str):
+        candidates = [raw_media]
+    elif isinstance(raw_media, list):
+        candidates = [str(item) for item in raw_media if isinstance(item, str) and item.strip()]
+    else:
+        candidates = []
+    for raw in candidates:
+        try:
+            path = Path(raw).expanduser()
+            if not path.is_absolute():
+                path = path.resolve()
+            seen.add(str(path))
+        except Exception:
+            continue
+
+
+def _extract_final_reply_media(reply: str, emitted_media: set[str]) -> list[str]:
+    """Return local image paths mentioned in final text that were not already emitted."""
+    media: list[str] = []
+    seen = set(emitted_media)
+    for match in _FINAL_REPLY_IMAGE_PATH_RE.finditer(reply or ""):
+        raw = match.group("path").strip(" \t\r\n\"'.,;:，。；：、)]}")
+        if not raw:
+            continue
+        path = Path(raw).expanduser()
+        if not path.is_absolute():
+            continue
+        if not path.is_file():
+            continue
+        resolved = str(path)
+        if resolved in seen:
+            continue
+        seen.add(resolved)
+        media.append(resolved)
+    return media
+
+
 def _format_tool_media_caption(event: ToolExecutionCompleted, media: list[str]) -> str:
     """Return a short caption for media generated by tools."""
     if event.tool_name == "image_generation":

ohmo/gateway/service.py

@@ -290,6 +290,7 @@ def start_gateway_process(cwd: str | Path | None = None, workspace: str | Path |
                 service._cwd,
                 "--workspace",
                 str(get_workspace_root(workspace)),
+                "--no-console-log",
             ],
             **popen_kwargs,
         )

src/openharness/cli.py

@@ -768,6 +768,7 @@ mcp_app = typer.Typer(name="mcp", help="Manage MCP servers")
 plugin_app = typer.Typer(name="plugin", help="Manage plugins")
 auth_app = typer.Typer(name="auth", help="Manage authentication")
 provider_app = typer.Typer(name="provider", help="Manage provider profiles")
+config_app = typer.Typer(name="config", help="Show or update settings")
 cron_app = typer.Typer(name="cron", help="Manage cron scheduler and jobs")
 autopilot_app = typer.Typer(name="autopilot", help="Manage repo autopilot")
 
@@ -775,6 +776,7 @@ app.add_typer(mcp_app)
 app.add_typer(plugin_app)
 app.add_typer(auth_app)
 app.add_typer(provider_app)
+app.add_typer(config_app)
 app.add_typer(cron_app)
 app.add_typer(autopilot_app)
 
@@ -1967,6 +1969,72 @@ def auth_copilot_logout() -> None:
     print("Copilot authentication cleared.")
 
 
+# ---- config subcommands ----
+
+
+def _config_resolve_target(settings: object, key: str) -> tuple[object, str]:
+    target = settings
+    parts = key.split(".")
+    for part in parts[:-1]:
+        if not hasattr(target, part):
+            raise KeyError(key)
+        target = getattr(target, part)
+    leaf = parts[-1]
+    if not hasattr(target, leaf):
+        raise KeyError(key)
+    return target, leaf
+
+
+def _config_coerce_value(current: object, raw: str) -> object:
+    if isinstance(current, bool):
+        lowered = raw.strip().lower()
+        if lowered in {"1", "true", "yes", "on"}:
+            return True
+        if lowered in {"0", "false", "no", "off"}:
+            return False
+        raise ValueError(f"Invalid boolean value: {raw}")
+    if isinstance(current, int) and not isinstance(current, bool):
+        return int(raw)
+    if isinstance(current, float):
+        return float(raw)
+    if isinstance(current, list):
+        return [entry.strip() for entry in raw.split(",") if entry.strip()]
+    return raw
+
+
+@config_app.command("show")
+def config_show() -> None:
+    """Print the resolved settings JSON."""
+    from openharness.commands.registry import _settings_json_for_display
+    from openharness.config.settings import load_settings
+
+    print(_settings_json_for_display(load_settings()), flush=True)
+
+
+@config_app.command("set")
+def config_set(
+    key: str = typer.Argument(..., help="Setting key, including dotted nested keys"),
+    value: str = typer.Argument(..., help="Value to store"),
+) -> None:
+    """Persist one setting in ~/.openharness/settings.json."""
+    from openharness.config.settings import load_settings, save_settings
+
+    settings = load_settings()
+    try:
+        target, leaf = _config_resolve_target(settings, key)
+    except KeyError:
+        print(f"Unknown config key: {key}", file=sys.stderr)
+        raise typer.Exit(1)
+    try:
+        coerced = _config_coerce_value(getattr(target, leaf), value)
+    except ValueError as exc:
+        print(f"Error: {exc}", file=sys.stderr)
+        raise typer.Exit(1)
+    setattr(target, leaf, coerced)
+    save_settings(settings)
+    print(f"Updated {key}", flush=True)
+
+
 # ---- provider subcommands ----
 
 

src/openharness/config/settings.py

@@ -113,6 +113,14 @@ class SandboxSettings(BaseModel):
     docker: DockerSandboxSettings = Field(default_factory=DockerSandboxSettings)
 
 
+class WebSettings(BaseModel):
+    """Outbound web tool configuration."""
+
+    proxy: str | None = None
+    resolution_mode: str = "auto"
+    synthetic_dns_cidrs: list[str] = Field(default_factory=list)
+
+
 class ProviderProfile(BaseModel):
     """Named provider workflow configuration."""
 
@@ -578,6 +586,7 @@ class Settings(BaseModel):
     hooks: dict[str, list[HookDefinition]] = Field(default_factory=dict)
     memory: MemorySettings = Field(default_factory=MemorySettings)
     sandbox: SandboxSettings = Field(default_factory=SandboxSettings)
+    web: WebSettings = Field(default_factory=WebSettings)
     enabled_plugins: dict[str, bool] = Field(default_factory=dict)
     allow_project_plugins: bool = False
     allow_project_skills: bool = True
@@ -860,20 +869,24 @@ class Settings(BaseModel):
         """Return a new Settings with CLI overrides applied (non-None values only)."""
         updates = {k: v for k, v in overrides.items() if v is not None}
         permission_mode = updates.pop("permission_mode", None)
+
+        def apply_permission_mode(settings: Settings) -> Settings:
+            if permission_mode is None:
+                return settings
+            return settings.model_copy(
+                update={
+                    "permission": settings.permission.model_copy(
+                        update={"mode": PermissionMode(str(permission_mode))}
+                    )
+                }
+            )
+
         # Strip ANSI escape sequences from model name if present
         if "model" in updates and isinstance(updates["model"], str):
             updates["model"] = strip_ansi_escape_sequences(updates["model"])
         if "effort" in updates and isinstance(updates["effort"], str):
             updates["effort"] = "xhigh" if updates["effort"].strip().lower() == "max" else updates["effort"].strip().lower()
-        merged = self.model_copy(update=updates)
-        if permission_mode is not None:
-            merged = merged.model_copy(
-                update={
-                    "permission": merged.permission.model_copy(
-                        update={"mode": PermissionMode(str(permission_mode))}
-                    )
-                }
-            )
+        merged = apply_permission_mode(self.model_copy(update=updates))
         if not updates:
             return merged
         profile_keys = {
@@ -890,8 +903,25 @@ class Settings(BaseModel):
         profile_updates = profile_keys.intersection(updates)
         if not profile_updates:
             return merged
-        if profile_updates.issubset({"active_profile"}):
-            return merged.materialize_active_profile()
+        if "active_profile" in profile_updates:
+            switch_updates = {
+                key: value
+                for key, value in updates.items()
+                if key not in profile_keys or key in {"active_profile", "profiles"}
+            }
+            switched = apply_permission_mode(self.model_copy(update=switch_updates)).materialize_active_profile()
+            remaining_profile_updates = {
+                key: value
+                for key, value in updates.items()
+                if key in profile_keys and key not in {"active_profile", "profiles"}
+            }
+            if not remaining_profile_updates:
+                return switched
+            return (
+                switched.model_copy(update=remaining_profile_updates)
+                .sync_active_profile_from_flat_fields()
+                .materialize_active_profile()
+            )
         return merged.sync_active_profile_from_flat_fields().materialize_active_profile()
 
 
@@ -987,6 +1017,23 @@ def _apply_env_overrides(settings: Settings) -> Settings:
     if sandbox_updates:
         updates["sandbox"] = settings.sandbox.model_copy(update=sandbox_updates)
 
+    web_updates: dict[str, Any] = {}
+    web_proxy = os.environ.get("OPENHARNESS_WEB_PROXY")
+    if web_proxy:
+        web_updates["proxy"] = web_proxy
+    web_resolution_mode = os.environ.get("OPENHARNESS_WEB_RESOLUTION_MODE")
+    if web_resolution_mode:
+        web_updates["resolution_mode"] = web_resolution_mode
+    web_synthetic_dns_cidrs = os.environ.get("OPENHARNESS_WEB_SYNTHETIC_DNS_CIDRS")
+    if web_synthetic_dns_cidrs:
+        web_updates["synthetic_dns_cidrs"] = [
+            entry.strip()
+            for entry in web_synthetic_dns_cidrs.split(",")
+            if entry.strip()
+        ]
+    if web_updates:
+        updates["web"] = settings.web.model_copy(update=web_updates)
+
     if not updates:
         return settings
     return settings.model_copy(update=updates)

src/openharness/utils/network_guard.py

@@ -4,9 +4,9 @@ from __future__ import annotations
 
 import asyncio
 import ipaddress
-import os
 import socket
-from urllib.parse import urljoin, urlparse
+from enum import Enum
+from urllib.parse import ParseResult, urljoin, urlparse
 
 import httpx
 
@@ -16,6 +16,31 @@ _DEFAULT_PORTS = {
     "https": 443,
 }
 _IPAddress = ipaddress.IPv4Address | ipaddress.IPv6Address
+_IPNetwork = ipaddress.IPv4Network | ipaddress.IPv6Network
+_SYNTHETIC_DNS_CIDRS_SETTING = "web.synthetic_dns_cidrs"
+_RESOLUTION_MODE_SETTING = "web.resolution_mode"
+_PROXY_SETTING = "web.proxy"
+_LOCAL_HOSTNAMES = {
+    "localhost",
+    "localhost.localdomain",
+    "metadata.google.internal",
+}
+_LOCAL_HOST_SUFFIXES = (
+    ".localhost",
+    ".local",
+    ".localdomain",
+    ".internal",
+    ".cluster.local",
+)
+
+
+class ResolutionMode(str, Enum):
+    """How outbound web tools should interpret target DNS resolution."""
+
+    AUTO = "auto"
+    DIRECT = "direct"
+    PROXY = "proxy"
+    SYNTHETIC_DNS = "synthetic_dns"
 
 
 class NetworkGuardError(ValueError):
@@ -33,22 +58,73 @@ def validate_http_url(url: str) -> None:
         raise NetworkGuardError("URLs with embedded credentials are not allowed")
 
 
+def get_web_resolution_mode(
+    proxy: str | None = None,
+    *,
+    configured_mode: str | None = None,
+) -> ResolutionMode:
+    """Resolve the configured web target validation mode."""
+    raw_mode = (configured_mode or "").strip().lower().replace("-", "_")
+    if not raw_mode or raw_mode == ResolutionMode.AUTO.value:
+        return ResolutionMode.PROXY if proxy else ResolutionMode.DIRECT
+    try:
+        mode = ResolutionMode(raw_mode)
+    except ValueError as exc:
+        allowed = ", ".join(mode.value for mode in ResolutionMode)
+        raise NetworkGuardError(f"{_RESOLUTION_MODE_SETTING} must be one of: {allowed}") from exc
+    if mode is ResolutionMode.AUTO:
+        return ResolutionMode.PROXY if proxy else ResolutionMode.DIRECT
+    if mode is ResolutionMode.PROXY and not proxy:
+        raise NetworkGuardError(f"{_RESOLUTION_MODE_SETTING}=proxy requires {_PROXY_SETTING}")
+    return mode
+
+
+def parse_synthetic_dns_cidrs(value: str | None = None) -> tuple[_IPNetwork, ...]:
+    """Parse user-declared synthetic DNS CIDRs."""
+    raw_value = "" if value is None else value
+    entries = [entry.strip() for entry in raw_value.split(",") if entry.strip()]
+    networks: list[_IPNetwork] = []
+    for entry in entries:
+        try:
+            networks.append(ipaddress.ip_network(entry, strict=False))
+        except ValueError as exc:
+            raise NetworkGuardError(f"invalid {_SYNTHETIC_DNS_CIDRS_SETTING} entry: {entry}") from exc
+    return tuple(networks)
+
+
 async def ensure_public_http_url(url: str) -> None:
     """Reject loopback, private-network, and other non-public HTTP targets."""
-    validate_http_url(url)
-    parsed = urlparse(url)
-    assert parsed.hostname is not None  # covered by validate_http_url
+    parsed = _validated_parsed_http_url(url)
+    hostname = _normalized_hostname(parsed.hostname)
+    literal = _parse_ip_literal(hostname)
+    if literal is not None:
+        _ensure_global_literal_ip(literal)
+        return
+    _ensure_not_local_hostname(hostname)
     port = parsed.port or _DEFAULT_PORTS[parsed.scheme]
-    addresses = await _resolve_host_addresses(parsed.hostname, port)
+    addresses = await _resolve_host_addresses(hostname, port)
     if not addresses:
-        raise NetworkGuardError(f"target host did not resolve: {parsed.hostname}")
+        raise NetworkGuardError(f"target host did not resolve: {hostname}")
 
     blocked = sorted({str(address) for address in addresses if not address.is_global})
     if blocked:
-        rendered = ", ".join(blocked[:3])
-        if len(blocked) > 3:
-            rendered += ", ..."
-        raise NetworkGuardError(f"target resolves to non-public address(es): {rendered}")
+        raise NetworkGuardError(_format_blocked_addresses(blocked, include_synthetic_dns_hint=True))
+
+
+async def ensure_http_url_allowed(
+    url: str,
+    *,
+    mode: ResolutionMode,
+    synthetic_cidrs: tuple[_IPNetwork, ...] = (),
+) -> None:
+    """Validate one outbound URL according to the configured resolution mode."""
+    if mode is ResolutionMode.DIRECT:
+        await ensure_public_http_url(url)
+        return
+    if mode is ResolutionMode.PROXY:
+        _ensure_proxy_safe_http_url(url)
+        return
+    await _ensure_synthetic_dns_safe_http_url(url, synthetic_cidrs=synthetic_cidrs)
 
 
 async def fetch_public_http_response(
@@ -64,9 +140,19 @@ async def fetch_public_http_response(
     current_url = url
     current_params = params
 
-    resolved_proxy = proxy if proxy is not None else os.environ.get("OPENHARNESS_WEB_PROXY")
+    web_settings = _load_configured_web_settings()
+    resolved_proxy = proxy if proxy is not None else web_settings.proxy
     if resolved_proxy:
         validate_http_url(resolved_proxy)
+    mode = get_web_resolution_mode(
+        resolved_proxy,
+        configured_mode=web_settings.resolution_mode,
+    )
+    synthetic_cidrs = (
+        parse_synthetic_dns_cidrs(",".join(web_settings.synthetic_dns_cidrs))
+        if mode is ResolutionMode.SYNTHETIC_DNS
+        else ()
+    )
 
     async with httpx.AsyncClient(
         follow_redirects=False,
@@ -75,7 +161,11 @@ async def fetch_public_http_response(
         proxy=resolved_proxy,
     ) as client:
         for redirect_count in range(max_redirects + 1):
-            await ensure_public_http_url(current_url)
+            await ensure_http_url_allowed(
+                current_url,
+                mode=mode,
+                synthetic_cidrs=synthetic_cidrs,
+            )
             response = await client.get(
                 current_url,
                 params=current_params,
@@ -96,6 +186,75 @@ async def fetch_public_http_response(
     raise NetworkGuardError("request failed before receiving a response")
 
 
+class _ConfiguredWebSettings:
+    def __init__(
+        self,
+        *,
+        proxy: str | None,
+        resolution_mode: str,
+        synthetic_dns_cidrs: list[str],
+    ) -> None:
+        self.proxy = proxy
+        self.resolution_mode = resolution_mode
+        self.synthetic_dns_cidrs = synthetic_dns_cidrs
+
+
+def _load_configured_web_settings() -> _ConfiguredWebSettings:
+    """Load persisted web settings, including environment overrides."""
+    from openharness.config import load_settings
+
+    web = load_settings().web
+    return _ConfiguredWebSettings(
+        proxy=web.proxy,
+        resolution_mode=web.resolution_mode,
+        synthetic_dns_cidrs=list(web.synthetic_dns_cidrs),
+    )
+
+
+def _ensure_proxy_safe_http_url(url: str) -> None:
+    """Validate a URL whose hostname will be resolved by an explicit proxy."""
+    parsed = _validated_parsed_http_url(url)
+    hostname = _normalized_hostname(parsed.hostname)
+    literal = _parse_ip_literal(hostname)
+    if literal is not None:
+        _ensure_global_literal_ip(literal)
+        return
+    _ensure_not_local_hostname(hostname)
+
+
+async def _ensure_synthetic_dns_safe_http_url(
+    url: str,
+    *,
+    synthetic_cidrs: tuple[_IPNetwork, ...],
+) -> None:
+    """Validate a URL in a user-declared synthetic DNS environment."""
+    if not synthetic_cidrs:
+        raise NetworkGuardError(
+            f"{ResolutionMode.SYNTHETIC_DNS.value} mode requires {_SYNTHETIC_DNS_CIDRS_SETTING}"
+        )
+    parsed = _validated_parsed_http_url(url)
+    hostname = _normalized_hostname(parsed.hostname)
+    literal = _parse_ip_literal(hostname)
+    if literal is not None:
+        _ensure_global_literal_ip(literal)
+        return
+    _ensure_not_local_hostname(hostname)
+    port = parsed.port or _DEFAULT_PORTS[parsed.scheme]
+    addresses = await _resolve_host_addresses(hostname, port)
+    if not addresses:
+        raise NetworkGuardError(f"target host did not resolve: {hostname}")
+
+    blocked = sorted(
+        {
+            str(address)
+            for address in addresses
+            if not address.is_global and not _address_in_networks(address, synthetic_cidrs)
+        }
+    )
+    if blocked:
+        raise NetworkGuardError(_format_blocked_addresses(blocked))
+
+
 async def _resolve_host_addresses(host: str, port: int) -> set[_IPAddress]:
     """Resolve a host into concrete IP addresses."""
     literal = _parse_ip_literal(host)
@@ -121,6 +280,8 @@ async def _resolve_host_addresses(host: str, port: int) -> set[_IPAddress]:
             candidate = sockaddr[0]
         else:
             continue
+        if not isinstance(candidate, str):
+            continue
         parsed = _parse_ip_literal(candidate)
         if parsed is not None:
             addresses.add(parsed)
@@ -132,3 +293,48 @@ def _parse_ip_literal(value: str) -> _IPAddress | None:
         return ipaddress.ip_address(value)
     except ValueError:
         return None
+
+
+def _validated_parsed_http_url(url: str) -> ParseResult:
+    validate_http_url(url)
+    parsed = urlparse(url)
+    assert parsed.hostname is not None  # covered by validate_http_url
+    return parsed
+
+
+def _normalized_hostname(hostname: str | None) -> str:
+    assert hostname is not None  # covered by validate_http_url
+    return hostname.rstrip(".").lower()
+
+
+def _ensure_global_literal_ip(address: _IPAddress) -> None:
+    if not address.is_global:
+        raise NetworkGuardError(f"target resolves to non-public address(es): {address}")
+
+
+def _ensure_not_local_hostname(hostname: str) -> None:
+    if hostname in _LOCAL_HOSTNAMES or any(hostname.endswith(suffix) for suffix in _LOCAL_HOST_SUFFIXES):
+        raise NetworkGuardError(f"local hostnames are not allowed: {hostname}")
+    if "." not in hostname:
+        raise NetworkGuardError(f"single-label hostnames are not allowed: {hostname}")
+
+
+def _address_in_networks(address: _IPAddress, networks: tuple[_IPNetwork, ...]) -> bool:
+    return any(address.version == network.version and address in network for network in networks)
+
+
+def _format_blocked_addresses(
+    blocked: list[str],
+    *,
+    include_synthetic_dns_hint: bool = False,
+) -> str:
+    rendered = ", ".join(blocked[:3])
+    if len(blocked) > 3:
+        rendered += ", ..."
+    message = f"target resolves to non-public address(es): {rendered}"
+    if include_synthetic_dns_hint:
+        message += (
+            "; if this domain intentionally resolves through synthetic DNS, configure "
+            "web.resolution_mode=synthetic_dns and web.synthetic_dns_cidrs=<cidr>"
+        )
+    return message

tests/test_config/test_settings.py

@@ -32,6 +32,8 @@ class TestSettings:
         assert s.permission.mode == "default"
         assert s.sandbox.enabled is False
         assert s.sandbox.filesystem.allow_write == ["."]
+        assert s.web.resolution_mode == "auto"
+        assert s.web.synthetic_dns_cidrs == []
 
     def test_resolve_api_key_from_instance(self):
         s = Settings(api_key="sk-test-123")
@@ -78,6 +80,17 @@ class TestSettings:
         assert updated.permission.mode == "full_auto"
         assert s.permission.mode == "default"
 
+    def test_web_settings_env_overrides(self, monkeypatch):
+        monkeypatch.setenv("OPENHARNESS_WEB_PROXY", "http://proxy.example.com:7890")
+        monkeypatch.setenv("OPENHARNESS_WEB_RESOLUTION_MODE", "synthetic_dns")
+        monkeypatch.setenv("OPENHARNESS_WEB_SYNTHETIC_DNS_CIDRS", "100.64.0.0/10,203.0.113.0/24")
+
+        updated = _apply_env_overrides(Settings())
+
+        assert updated.web.proxy == "http://proxy.example.com:7890"
+        assert updated.web.resolution_mode == "synthetic_dns"
+        assert updated.web.synthetic_dns_cidrs == ["100.64.0.0/10", "203.0.113.0/24"]
+
     def test_merge_cli_overrides_returns_new_instance(self):
         s = Settings()
         updated = s.merge_cli_overrides(model="claude-opus-4-20250514")
@@ -330,6 +343,46 @@ class TestLoadSaveSettings:
         assert profile.provider == "openai_codex"
         assert profile.auth_source == "codex_subscription"
 
+    def test_merge_cli_active_profile_with_model_keeps_target_profile_auth(self):
+        settings = Settings(
+            active_profile="moonshot",
+            provider="moonshot",
+            api_format="openai",
+            base_url="https://api.moonshot.cn/v1",
+            model="kimi-k2.5",
+            profiles={
+                "moonshot": ProviderProfile(
+                    label="Moonshot",
+                    provider="moonshot",
+                    api_format="openai",
+                    auth_source="moonshot_api_key",
+                    default_model="kimi-k2.5",
+                    last_model="kimi-k2.5",
+                    base_url="https://api.moonshot.cn/v1",
+                ),
+                "codex": ProviderProfile(
+                    label="Codex Subscription",
+                    provider="openai_codex",
+                    api_format="openai",
+                    auth_source="codex_subscription",
+                    default_model="gpt-5.4",
+                    last_model="gpt-5.4",
+                ),
+            },
+        )
+
+        updated = settings.merge_cli_overrides(active_profile="codex", model="gpt-5.5")
+        profile_name, profile = updated.resolve_profile()
+
+        assert profile_name == "codex"
+        assert updated.provider == "openai_codex"
+        assert updated.api_format == "openai"
+        assert updated.base_url is None
+        assert updated.model == "gpt-5.5"
+        assert profile.provider == "openai_codex"
+        assert profile.auth_source == "codex_subscription"
+        assert profile.last_model == "gpt-5.5"
+
     def test_merge_cli_active_profile_keeps_profile_compact_threshold_settings(self):
         settings = Settings(
             active_profile="moonshot",

tests/test_entrypoints/test_config_cli.py

@@ -0,0 +1,30 @@
+"""Tests for the top-level config CLI."""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+from typer.testing import CliRunner
+
+from openharness.cli import app
+from openharness.config.settings import load_settings
+
+
+def test_cli_config_set_persists_nested_web_settings(tmp_path: Path, monkeypatch):
+    monkeypatch.setenv("OPENHARNESS_CONFIG_DIR", str(tmp_path / "config"))
+    runner = CliRunner()
+
+    mode_result = runner.invoke(app, ["config", "set", "web.resolution_mode", "synthetic_dns"])
+    assert mode_result.exit_code == 0
+    assert "Updated web.resolution_mode" in mode_result.output
+
+    cidrs_result = runner.invoke(
+        app,
+        ["config", "set", "web.synthetic_dns_cidrs", "100.64.0.0/10,203.0.113.0/24"],
+    )
+    assert cidrs_result.exit_code == 0
+    assert "Updated web.synthetic_dns_cidrs" in cidrs_result.output
+
+    settings = load_settings()
+    assert settings.web.resolution_mode == "synthetic_dns"
+    assert settings.web.synthetic_dns_cidrs == ["100.64.0.0/10", "203.0.113.0/24"]

tests/test_ohmo/test_cli.py

@@ -1,9 +1,10 @@
 import json
+import logging
 from pathlib import Path
 
 from typer.testing import CliRunner
 
-from ohmo.cli import app
+from ohmo.cli import _build_gateway_logging_handlers, app
 
 
 def test_ohmo_help():
@@ -48,6 +49,42 @@ def test_ohmo_init_noninteractive_defaults_to_deny_all_remote_access(tmp_path: P
     assert config["channel_configs"] == {}
 
 
+def test_gateway_logging_handlers_write_gateway_log_file(tmp_path: Path):
+    workspace = tmp_path / ".ohmo-home"
+    runner = CliRunner()
+    result = runner.invoke(app, ["init", "--workspace", str(workspace), "--no-interactive"])
+    assert result.exit_code == 0
+
+    handlers = _build_gateway_logging_handlers(workspace, console=True, log_file=True)
+    try:
+        file_handlers = [handler for handler in handlers if isinstance(handler, logging.FileHandler)]
+        console_handlers = [
+            handler
+            for handler in handlers
+            if isinstance(handler, logging.StreamHandler) and not isinstance(handler, logging.FileHandler)
+        ]
+        assert len(file_handlers) == 1
+        assert len(console_handlers) == 1
+
+        record = logging.LogRecord(
+            name="ohmo.gateway.test",
+            level=logging.INFO,
+            pathname=__file__,
+            lineno=1,
+            msg="GATEWAY_LOG_OK",
+            args=(),
+            exc_info=None,
+        )
+        file_handlers[0].emit(record)
+        file_handlers[0].flush()
+
+        log_path = workspace / "logs" / "gateway.log"
+        assert "GATEWAY_LOG_OK" in log_path.read_text(encoding="utf-8")
+    finally:
+        for handler in handlers:
+            handler.close()
+
+
 def test_ohmo_init_interactive_writes_gateway_config(tmp_path: Path, monkeypatch):
     runner = CliRunner()
     workspace = tmp_path / ".ohmo-home"

tests/test_ohmo/test_gateway.py

@@ -2,6 +2,7 @@ import asyncio
 import contextlib
 import logging
 import subprocess
+import sys
 from types import SimpleNamespace
 from datetime import datetime
 import json
@@ -46,7 +47,7 @@ from ohmo.gateway.runtime import (
     _sanitize_group_command_metadata,
     _sanitize_group_command_prompts,
 )
-from ohmo.gateway.service import OhmoGatewayService, gateway_status, stop_gateway_process
+from ohmo.gateway.service import OhmoGatewayService, gateway_status, start_gateway_process, stop_gateway_process
 from ohmo.group_registry import load_managed_group_record, save_managed_group_record
 from ohmo.memory import add_memory_entry as add_ohmo_memory_entry
 from ohmo.memory import list_memory_files as list_ohmo_memory_files
@@ -423,6 +424,34 @@ def test_gateway_status_prefers_live_config_over_stale_state(tmp_path):
     assert state.enabled_channels == ["feishu"]
 
 
+def test_start_gateway_process_uses_child_log_file_handler_without_console_duplication(tmp_path, monkeypatch):
+    workspace = tmp_path / ".ohmo-home"
+    initialize_workspace(workspace)
+    captured: dict[str, object] = {}
+
+    class FakeProcess:
+        pid = 1234
+
+    def fake_popen(args, **kwargs):
+        captured["args"] = args
+        captured["kwargs"] = kwargs
+        return FakeProcess()
+
+    monkeypatch.setattr("ohmo.gateway.service.subprocess.Popen", fake_popen)
+
+    assert start_gateway_process(tmp_path, workspace) == 1234
+
+    args = captured["args"]
+    kwargs = captured["kwargs"]
+    assert isinstance(args, list)
+    assert args[:4] == [sys.executable, "-m", "ohmo", "gateway"]
+    assert "run" in args
+    assert "--no-console-log" in args
+    assert isinstance(kwargs, dict)
+    assert kwargs["stdout"] is kwargs["stderr"]
+    assert getattr(kwargs["stdout"], "name", "").endswith("gateway.log")
+
+
 def test_stop_gateway_process_kills_matching_workspace_processes(tmp_path, monkeypatch):
     workspace = tmp_path / ".ohmo-home"
     workspace.mkdir()
@@ -770,6 +799,95 @@ async def test_runtime_pool_stream_message_emits_media_for_generated_tool_paths(
     assert "已生成图片 via codex" in media_updates[0].text
 
 
+@pytest.mark.asyncio
+async def test_runtime_pool_attaches_final_reply_image_path_as_media(tmp_path, monkeypatch):
+    workspace = tmp_path / ".ohmo-home"
+    initialize_workspace(workspace)
+    image_path = tmp_path / "generated.png"
+    image_path.write_bytes(b"png")
+
+    async def fake_build_runtime(**kwargs):
+        class FakeEngine:
+            messages = []
+            total_usage = UsageSnapshot()
+
+            def set_system_prompt(self, prompt):
+                return None
+
+            async def submit_message(self, content):
+                yield AssistantTextDelta(text=f"已生成图片：\n```text\n{image_path}\n```")
+
+        return SimpleNamespace(
+            engine=FakeEngine(),
+            cwd=str(tmp_path),
+            session_id="sess123",
+            current_settings=lambda: SimpleNamespace(model="gpt-5.4"),
+            commands=SimpleNamespace(lookup=lambda raw: None),
+        )
+
+    async def fake_start_runtime(bundle):
+        return None
+
+    monkeypatch.setattr("ohmo.gateway.runtime.build_runtime", fake_build_runtime)
+    monkeypatch.setattr("ohmo.gateway.runtime.start_runtime", fake_start_runtime)
+
+    pool = OhmoSessionRuntimePool(cwd=tmp_path, workspace=workspace, provider_profile="codex")
+    message = InboundMessage(channel="feishu", sender_id="u1", chat_id="c1", content="draw")
+    updates = [u async for u in pool.stream_message(message, "feishu:c1")]
+
+    assert updates[-1].kind == "final"
+    assert updates[-1].media == [str(image_path)]
+    assert updates[-1].metadata["_media"] == [str(image_path)]
+    assert updates[-1].metadata["_final_media_fallback"] is True
+
+
+@pytest.mark.asyncio
+async def test_runtime_pool_does_not_duplicate_final_reply_image_media(tmp_path, monkeypatch):
+    workspace = tmp_path / ".ohmo-home"
+    initialize_workspace(workspace)
+    image_path = tmp_path / "generated.png"
+    image_path.write_bytes(b"png")
+
+    async def fake_build_runtime(**kwargs):
+        class FakeEngine:
+            messages = []
+            total_usage = UsageSnapshot()
+
+            def set_system_prompt(self, prompt):
+                return None
+
+            async def submit_message(self, content):
+                yield ToolExecutionCompleted(
+                    tool_name="image_generation",
+                    output=f"Wrote {image_path}",
+                    metadata={"paths": [str(image_path)], "provider": "codex"},
+                )
+                yield AssistantTextDelta(text=f"已生成图片：\n```text\n{image_path}\n```")
+
+        return SimpleNamespace(
+            engine=FakeEngine(),
+            cwd=str(tmp_path),
+            session_id="sess123",
+            current_settings=lambda: SimpleNamespace(model="gpt-5.4"),
+            commands=SimpleNamespace(lookup=lambda raw: None),
+        )
+
+    async def fake_start_runtime(bundle):
+        return None
+
+    monkeypatch.setattr("ohmo.gateway.runtime.build_runtime", fake_build_runtime)
+    monkeypatch.setattr("ohmo.gateway.runtime.start_runtime", fake_start_runtime)
+
+    pool = OhmoSessionRuntimePool(cwd=tmp_path, workspace=workspace, provider_profile="codex")
+    message = InboundMessage(channel="feishu", sender_id="u1", chat_id="c1", content="draw")
+    updates = [u async for u in pool.stream_message(message, "feishu:c1")]
+
+    assert [u.kind for u in updates].count("media") == 1
+    assert updates[-1].kind == "final"
+    assert updates[-1].media is None
+    assert "_final_media_fallback" not in updates[-1].metadata
+
+
 @pytest.mark.asyncio
 async def test_runtime_pool_stream_message_formats_auto_compact_status_for_feishu(tmp_path, monkeypatch):
     workspace = tmp_path / ".ohmo-home"
@@ -1696,6 +1814,36 @@ async def test_gateway_bridge_publishes_media_updates():
     assert outbound.media == ["/tmp/generated.png"]
 
 
+@pytest.mark.asyncio
+async def test_gateway_bridge_publishes_final_media_updates():
+    bus = MessageBus()
+
+    class FakeRuntimePool:
+        async def stream_message(self, message, session_key):
+            yield SimpleNamespace(
+                kind="final",
+                text="已生成图片：generated.png",
+                media=["/tmp/generated.png"],
+                metadata={"_session_key": session_key, "_media": ["/tmp/generated.png"]},
+            )
+
+    bridge = OhmoGatewayBridge(bus=bus, runtime_pool=FakeRuntimePool())
+    task = asyncio.create_task(bridge.run())
+    try:
+        await bus.publish_inbound(InboundMessage(channel="feishu", sender_id="u1", chat_id="c1", content="draw"))
+        outbound = await asyncio.wait_for(bus.consume_outbound(), timeout=1.0)
+    finally:
+        bridge.stop()
+        task.cancel()
+        try:
+            await task
+        except asyncio.CancelledError:
+            pass
+
+    assert outbound.content == "已生成图片：generated.png"
+    assert outbound.media == ["/tmp/generated.png"]
+
+
 @pytest.mark.asyncio
 async def test_gateway_bridge_publishes_progress_updates():
     bus = MessageBus()
@@ -2654,7 +2802,14 @@ def test_runtime_pool_sanitizes_internal_group_prompt_metadata():
 async def test_runtime_pool_provider_command_refresh_uses_gateway_profile(tmp_path, monkeypatch):
     workspace = tmp_path / ".ohmo-home"
     initialize_workspace(workspace)
-    save_gateway_config(GatewayConfig(provider_profile="kimi-anthropic"), workspace)
+    save_gateway_config(
+        GatewayConfig(
+            provider_profile="kimi-anthropic",
+            allow_remote_admin_commands=True,
+            allowed_remote_admin_commands=["provider", "model"],
+        ),
+        workspace,
+    )
     build_calls: list[dict[str, object]] = []
 
     statuses = {
@@ -2732,6 +2887,92 @@ async def test_runtime_pool_provider_command_refresh_uses_gateway_profile(tmp_pa
     assert build_calls[1]["active_profile"] == "codex"
 
 
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "command_text,command_name",
+    [("/provider codex", "provider"), ("/model gpt-5.5", "model")],
+)
+async def test_runtime_pool_rejects_gateway_scoped_command_without_admin_opt_in(
+    tmp_path,
+    monkeypatch,
+    command_text,
+    command_name,
+):
+    workspace = tmp_path / ".ohmo-home"
+    initialize_workspace(workspace)
+    save_gateway_config(GatewayConfig(provider_profile="kimi-anthropic"), workspace)
+    build_calls: list[dict[str, object]] = []
+    handler_invocations: list[tuple[str, str]] = []
+
+    def fake_provider_handler(args, **_kwargs):
+        handler_invocations.append(("provider", args))
+        return ("provider switched", True)
+
+    def fake_model_handler(args, **_kwargs):
+        handler_invocations.append(("model", args))
+        return ("model switched", True)
+
+    class FakeEngine:
+        def __init__(self):
+            self.messages = [ConversationMessage.from_user_text("before")]
+            self.total_usage = UsageSnapshot()
+
+        def set_system_prompt(self, prompt):
+            del prompt
+
+        async def submit_message(self, content):
+            del content
+            if False:
+                yield None
+
+    async def fake_build_runtime(**kwargs):
+        build_calls.append(kwargs)
+        return SimpleNamespace(
+            engine=FakeEngine(),
+            session_id="sess123",
+            current_settings=lambda: SimpleNamespace(model="gpt-5.4"),
+            commands=create_default_command_registry(),
+            hook_summary=lambda: "",
+            mcp_summary=lambda: "",
+            plugin_summary=lambda: "",
+            cwd=str(tmp_path),
+            tool_registry=None,
+            app_state=None,
+            session_backend=None,
+            extra_skill_dirs=(),
+            extra_plugin_roots=(),
+            enforce_max_turns=False,
+        )
+
+    async def fake_start_runtime(bundle):
+        del bundle
+
+    async def fake_close_runtime(bundle):
+        del bundle
+
+    monkeypatch.setattr(
+        "ohmo.gateway.runtime.handle_gateway_provider_command", fake_provider_handler
+    )
+    monkeypatch.setattr(
+        "ohmo.gateway.runtime.handle_gateway_model_command", fake_model_handler
+    )
+    monkeypatch.setattr("ohmo.gateway.runtime.build_runtime", fake_build_runtime)
+    monkeypatch.setattr("ohmo.gateway.runtime.start_runtime", fake_start_runtime)
+    monkeypatch.setattr("ohmo.gateway.runtime.close_runtime", fake_close_runtime)
+
+    pool = OhmoSessionRuntimePool(cwd=tmp_path, workspace=workspace, provider_profile="kimi-anthropic")
+    message = InboundMessage(channel="feishu", sender_id="u1", chat_id="c1", content=command_text)
+    updates = [u async for u in pool.stream_message(message, "feishu:c1")]
+
+    assert handler_invocations == []
+    assert any(
+        f"/{command_name} is only available in the local OpenHarness UI." in update.text
+        for update in updates
+    )
+    assert load_gateway_config(workspace).provider_profile == "kimi-anthropic"
+    assert len(build_calls) == 1
+
+
 @pytest.mark.asyncio
 async def test_runtime_pool_stream_message_handles_slash_command_and_refresh_runtime(tmp_path, monkeypatch):
     workspace = tmp_path / ".ohmo-home"

tests/test_utils/test_network_guard.py

@@ -0,0 +1,154 @@
+"""Tests for outbound HTTP target validation."""
+
+from __future__ import annotations
+
+import ipaddress
+
+import httpx
+import pytest
+
+from openharness.utils.network_guard import fetch_public_http_response
+
+
+class FakeAsyncClient:
+    def __init__(self, **kwargs: object) -> None:
+        self.kwargs = kwargs
+
+    async def __aenter__(self):
+        return self
+
+    async def __aexit__(self, exc_type, exc, tb):
+        return None
+
+    async def get(self, url: str, **kwargs: object) -> httpx.Response:
+        request = httpx.Request("GET", url, params=kwargs.get("params"))
+        return httpx.Response(200, text="ok", request=request)
+
+
+@pytest.fixture(autouse=True)
+def isolated_config_dir(tmp_path, monkeypatch):
+    monkeypatch.setenv("OPENHARNESS_CONFIG_DIR", str(tmp_path / "config"))
+
+
+@pytest.mark.asyncio
+async def test_fetch_public_http_response_direct_rejects_non_public_dns(monkeypatch):
+    async def fake_resolve(host: str, port: int):
+        return {ipaddress.ip_address("100.64.1.2")}
+
+    monkeypatch.setattr("openharness.utils.network_guard._resolve_host_addresses", fake_resolve)
+
+    with pytest.raises(ValueError, match="synthetic DNS"):
+        await fetch_public_http_response("https://example.com/")
+
+
+@pytest.mark.asyncio
+async def test_fetch_public_http_response_synthetic_dns_allows_declared_cidr(monkeypatch):
+    async def fake_resolve(host: str, port: int):
+        return {ipaddress.ip_address("100.64.1.2")}
+
+    monkeypatch.setenv("OPENHARNESS_WEB_RESOLUTION_MODE", "synthetic_dns")
+    monkeypatch.setenv("OPENHARNESS_WEB_SYNTHETIC_DNS_CIDRS", "100.64.0.0/10")
+    monkeypatch.setattr("openharness.utils.network_guard._resolve_host_addresses", fake_resolve)
+    monkeypatch.setattr(httpx, "AsyncClient", FakeAsyncClient)
+
+    response = await fetch_public_http_response("https://example.com/")
+
+    assert response.status_code == 200
+
+
+@pytest.mark.asyncio
+async def test_fetch_public_http_response_synthetic_dns_uses_persisted_settings(
+    monkeypatch,
+):
+    from openharness.config.settings import Settings, WebSettings, save_settings
+
+    async def fake_resolve(host: str, port: int):
+        return {ipaddress.ip_address("100.64.1.2")}
+
+    save_settings(
+        Settings(
+            web=WebSettings(
+                resolution_mode="synthetic_dns",
+                synthetic_dns_cidrs=["100.64.0.0/10"],
+            )
+        )
+    )
+    monkeypatch.setattr("openharness.utils.network_guard._resolve_host_addresses", fake_resolve)
+    monkeypatch.setattr(httpx, "AsyncClient", FakeAsyncClient)
+
+    response = await fetch_public_http_response("https://example.com/")
+
+    assert response.status_code == 200
+
+
+@pytest.mark.asyncio
+async def test_fetch_public_http_response_synthetic_dns_requires_declared_cidrs(monkeypatch):
+    monkeypatch.setenv("OPENHARNESS_WEB_RESOLUTION_MODE", "synthetic_dns")
+
+    with pytest.raises(ValueError, match="web.synthetic_dns_cidrs"):
+        await fetch_public_http_response("https://example.com/")
+
+
+@pytest.mark.asyncio
+async def test_fetch_public_http_response_synthetic_dns_rejects_literal_non_public_ip(
+    monkeypatch,
+):
+    monkeypatch.setenv("OPENHARNESS_WEB_RESOLUTION_MODE", "synthetic_dns")
+    monkeypatch.setenv("OPENHARNESS_WEB_SYNTHETIC_DNS_CIDRS", "100.64.0.0/10")
+
+    with pytest.raises(ValueError, match="non-public"):
+        await fetch_public_http_response("http://100.64.1.2/")
+
+
+@pytest.mark.asyncio
+async def test_fetch_public_http_response_synthetic_dns_rejects_undeclared_private_dns(
+    monkeypatch,
+):
+    async def fake_resolve(host: str, port: int):
+        return {ipaddress.ip_address("10.0.0.1")}
+
+    monkeypatch.setenv("OPENHARNESS_WEB_RESOLUTION_MODE", "synthetic_dns")
+    monkeypatch.setenv("OPENHARNESS_WEB_SYNTHETIC_DNS_CIDRS", "100.64.0.0/10")
+    monkeypatch.setattr("openharness.utils.network_guard._resolve_host_addresses", fake_resolve)
+
+    with pytest.raises(ValueError, match="non-public"):
+        await fetch_public_http_response("https://example.com/")
+
+
+@pytest.mark.asyncio
+async def test_fetch_public_http_response_proxy_mode_does_not_resolve_target_dns(monkeypatch):
+    async def fail_resolve(host: str, port: int):
+        raise AssertionError("proxy mode should not resolve ordinary target domains locally")
+
+    monkeypatch.setenv("OPENHARNESS_WEB_PROXY", "http://proxy.example.com:7890")
+    monkeypatch.setenv("OPENHARNESS_WEB_SYNTHETIC_DNS_CIDRS", "not-a-cidr")
+    monkeypatch.setattr("openharness.utils.network_guard._resolve_host_addresses", fail_resolve)
+    monkeypatch.setattr(httpx, "AsyncClient", FakeAsyncClient)
+
+    response = await fetch_public_http_response("https://example.com/")
+
+    assert response.status_code == 200
+
+
+@pytest.mark.asyncio
+async def test_fetch_public_http_response_proxy_mode_rejects_literal_non_public_ip(monkeypatch):
+    monkeypatch.setenv("OPENHARNESS_WEB_PROXY", "http://proxy.example.com:7890")
+
+    with pytest.raises(ValueError, match="non-public"):
+        await fetch_public_http_response("http://127.0.0.1/")
+
+
+@pytest.mark.asyncio
+async def test_fetch_public_http_response_rejects_non_public_redirect_in_proxy_mode(
+    monkeypatch,
+):
+    class RedirectClient(FakeAsyncClient):
+        async def get(self, url: str, **kwargs: object) -> httpx.Response:
+            request = httpx.Request("GET", url, params=kwargs.get("params"))
+            return httpx.Response(302, headers={"Location": "http://127.0.0.1/"}, request=request)
+
+    monkeypatch.setenv("OPENHARNESS_WEB_PROXY", "http://proxy.example.com:7890")
+    monkeypatch.setattr(httpx, "AsyncClient", RedirectClient)
+
+    with pytest.raises(ValueError, match="non-public"):
+        await fetch_public_http_response("https://example.com/")