diff --git a/.env.example b/.env.example
index 47f2c3b..c246ff7 100644
--- a/.env.example
+++ b/.env.example
@@ -57,10 +57,10 @@ SPLUNK_TOKEN=
 # VCT_SPLUNK_CONFIG=
 
 # --- Splunk Cloud (ACS) ------------------------------------------------------
-# Read-only this release. The backend is deduced from SPLUNK_URL: on a
-# *.splunkcloud.com host, supported reads route via the ACS API automatically
-# (there is no flag or variable to pick a backend). `splunk inspect` reports
-# what the deduced backend supports.
+# The backend is deduced from SPLUNK_URL: on a *.splunkcloud.com host,
+# supported reads route via the ACS API automatically (there is no flag or
+# variable to pick a backend). `splunk inspect` reports what the deduced
+# backend supports.
 
 # ACS authentication token (Bearer). The stack name is derived from SPLUNK_URL;
 # set SPLUNK_ACS_STACK only to override it.
@@ -71,6 +71,21 @@ SPLUNK_TOKEN=
 # FedRAMP stacks use https://admin.splunkcloudgc.com.
 # SPLUNK_ACS_BASE_URL=
 
+# Cloud writes are opt-in and narrow: create/update/delete for index, role, and
+# hec-token only (never a CLI flag, so a saved command line cannot enable one
+# by accident). Everything else on Cloud -- including enable/disable on those
+# same three resources -- stays refused regardless.
+# SPLUNK_CLOUD_WRITE=true
+
+# Optional: a separate ACS token scoped to writes only. Falls back to
+# SPLUNK_ACS_TOKEN when unset; reads always use SPLUNK_ACS_TOKEN, never this one.
+# SPLUNK_ACS_WRITE_TOKEN=
+
+# Hide the Cloud stack name at untrusted output boundaries (e.g. CI logs) in
+# the target shown by prompts, JSON metadata, and error text. The audit log is
+# unaffected -- it always records the real host.
+# VCT_SPLUNK_REDACT_TARGET=1
+
 # --- Live test opt-ins -------------------------------------------------------
 
 # Enables live read tests. Enterprise writes also require SPLUNK_WRITE_TEST=true.
diff --git a/.github/scripts/scan-cloud-ci-leaks.py b/.github/scripts/scan-cloud-ci-leaks.py
new file mode 100644
index 0000000..6f9f6f2
--- /dev/null
+++ b/.github/scripts/scan-cloud-ci-leaks.py
@@ -0,0 +1,53 @@
+#!/usr/bin/env python3
+"""Fail when Cloud CI artifacts appear to contain targets or credentials."""
+
+from __future__ import annotations
+
+import os
+import re
+import sys
+from pathlib import Path
+
+_ENV_NAMES = (
+    "SPLUNK_URL",
+    "SPLUNK_ACS_STACK",
+    "SPLUNK_ACS_BASE_URL",
+    "SPLUNK_ACS_TOKEN",
+    "SPLUNK_TOKEN",
+)
+_PATTERNS = (
+    re.compile(r"Bearer "),
+    re.compile(r"eyJ[A-Za-z0-9_-]+\.[A-Za-z0-9_-]+\.[A-Za-z0-9_-]+"),
+    re.compile(r"admin\.splunk\.com/(?!<redacted>(?:/|$))[^/\s]+", re.IGNORECASE),
+    re.compile(r"(?<![A-Za-z0-9-])(?!<redacted>\.)[A-Za-z0-9-]+\.splunkcloud\.com", re.IGNORECASE),
+)
+
+
+def _count_matches(line: str, literals: tuple[str, ...]) -> int:
+    """Count leak signatures in one line without retaining their values."""
+    return sum(line.count(value) for value in literals) + sum(
+        len(pattern.findall(line)) for pattern in _PATTERNS
+    )
+
+
+def main(argv: list[str]) -> int:
+    """Scan each requested artifact and return nonzero when a leak is found."""
+    literals = tuple(value for name in _ENV_NAMES if len(value := os.environ.get(name, "")) >= 4)
+    found = False
+    for name in argv:
+        path = Path(name)
+        try:
+            lines = path.read_text(encoding="utf-8", errors="replace").splitlines()
+        except FileNotFoundError:
+            print(f"{path}:missing", file=sys.stderr)
+            continue
+        for line_number, line in enumerate(lines, start=1):
+            count = _count_matches(line, literals)
+            if count:
+                print(f"{path}:{line_number}:{count}", file=sys.stderr)
+                found = True
+    return int(found)
+
+
+if __name__ == "__main__":
+    raise SystemExit(main(sys.argv[1:]))
diff --git a/.github/workflows/cloud-read.yml b/.github/workflows/cloud-read.yml
index a34636d..83a523b 100644
--- a/.github/workflows/cloud-read.yml
+++ b/.github/workflows/cloud-read.yml
@@ -20,16 +20,6 @@ jobs:
     name: Cloud / ACS reads
     runs-on: ubuntu-latest
     timeout-minutes: 10
-    env:
-      SPLUNK_ACS_LIVE_TEST: "true"
-      SPLUNK_URL: ${{ secrets.SPLUNK_URL }}
-      SPLUNK_ACS_TOKEN: ${{ secrets.SPLUNK_ACS_TOKEN }}
-      SPLUNK_ACS_STACK: ${{ secrets.SPLUNK_ACS_STACK }}
-      SPLUNK_ACS_BASE_URL: ${{ secrets.SPLUNK_ACS_BASE_URL }}
-      # splunkd credential. Without it, every read Cloud does not serve stops at
-      # the credential check instead of reaching the dispatch layer, so the run
-      # covers far less than it appears to. The guard below says so out loud.
-      SPLUNK_TOKEN: ${{ secrets.SPLUNK_TOKEN }}
     steps:
       - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7
         with:
@@ -39,6 +29,10 @@ jobs:
       # step that reports what can be certified rather than a skipped job.
       - name: Check for a configured Cloud stack
         id: stack
+        env:
+          SPLUNK_URL: ${{ secrets.SPLUNK_URL }}
+          SPLUNK_ACS_TOKEN: ${{ secrets.SPLUNK_ACS_TOKEN }}
+          SPLUNK_TOKEN: ${{ secrets.SPLUNK_TOKEN }}
         run: bash .github/scripts/detect-cloud-stack.sh
 
       - if: steps.stack.outputs.ready == 'true'
@@ -56,28 +50,58 @@ jobs:
 
       - name: Cloud reads (every catalogued read command)
         if: steps.stack.outputs.ready == 'true'
-        run: >-
-          .venv/bin/pytest tests/integration/cloud/read/test_catalog.py
-          -m "integration and cloud and read"
-          -vv --color=yes --tb=short --junitxml=cloud-read.xml
+        env:
+          SPLUNK_ACS_LIVE_TEST: "true"
+          SPLUNK_URL: ${{ secrets.SPLUNK_URL }}
+          SPLUNK_ACS_TOKEN: ${{ secrets.SPLUNK_ACS_TOKEN }}
+          SPLUNK_ACS_STACK: ${{ secrets.SPLUNK_ACS_STACK }}
+          SPLUNK_ACS_BASE_URL: ${{ secrets.SPLUNK_ACS_BASE_URL }}
+          SPLUNK_TOKEN: ${{ secrets.SPLUNK_TOKEN }}
+          VCT_SPLUNK_REDACT_TARGET: "1"
+          VCT_SPLUNK_AUDIT: ${{ runner.temp }}/vct-splunk-audit.log
+        run: |
+          set -o pipefail
+          set +e
+          .venv/bin/pytest tests/integration/cloud/read/test_catalog.py \
+          -m "integration and cloud and read" \
+          -q --tb=line -r N \
+          -o addopts='--strict-markers --import-mode=importlib' \
+          --junitxml=cloud-read.xml \
+          | tee cloud-read.log
+          pytest_status=${PIPESTATUS[0]}
+          set -e
+          scan_files=(cloud-read.log cloud-acs.log cloud-read.xml cloud-acs.xml)
+          if [ -n "${GITHUB_STEP_SUMMARY:-}" ]; then
+            scan_files+=("$GITHUB_STEP_SUMMARY")
+          fi
+          python .github/scripts/scan-cloud-ci-leaks.py "${scan_files[@]}"
+          exit "$pytest_status"
 
       - name: Cloud ACS operations (below the CLI)
         if: steps.stack.outputs.ready == 'true'
-        run: >-
-          .venv/bin/pytest tests/integration/cloud/read/test_acs_operations.py
-          -m "integration and cloud and read"
-          -vv --color=yes --tb=short --junitxml=cloud-acs.xml
-
-      - name: Publish test summary
-        if: always() && steps.stack.outputs.ready == 'true'
-        uses: test-summary/action@37b508cfee6d4d080eedd00b5bb240a6a784a6a5 # v2
-        with:
-          paths: cloud-*.xml
-
-      - name: Upload test report
-        if: always() && steps.stack.outputs.ready == 'true'
-        uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7
-        with:
-          name: cloud-read-${{ github.run_id }}
-          path: cloud-*.xml
-          if-no-files-found: ignore
+        env:
+          SPLUNK_ACS_LIVE_TEST: "true"
+          SPLUNK_URL: ${{ secrets.SPLUNK_URL }}
+          SPLUNK_ACS_TOKEN: ${{ secrets.SPLUNK_ACS_TOKEN }}
+          SPLUNK_ACS_STACK: ${{ secrets.SPLUNK_ACS_STACK }}
+          SPLUNK_ACS_BASE_URL: ${{ secrets.SPLUNK_ACS_BASE_URL }}
+          SPLUNK_TOKEN: ${{ secrets.SPLUNK_TOKEN }}
+          VCT_SPLUNK_REDACT_TARGET: "1"
+          VCT_SPLUNK_AUDIT: ${{ runner.temp }}/vct-splunk-audit.log
+        run: |
+          set -o pipefail
+          set +e
+          .venv/bin/pytest tests/integration/cloud/read/test_acs_operations.py \
+          -m "integration and cloud and read" \
+          -q --tb=line -r N \
+          -o addopts='--strict-markers --import-mode=importlib' \
+          --junitxml=cloud-acs.xml \
+          | tee cloud-acs.log
+          pytest_status=${PIPESTATUS[0]}
+          set -e
+          scan_files=(cloud-read.log cloud-acs.log cloud-read.xml cloud-acs.xml)
+          if [ -n "${GITHUB_STEP_SUMMARY:-}" ]; then
+            scan_files+=("$GITHUB_STEP_SUMMARY")
+          fi
+          python .github/scripts/scan-cloud-ci-leaks.py "${scan_files[@]}"
+          exit "$pytest_status"
diff --git a/.github/workflows/cloud-write.yml b/.github/workflows/cloud-write.yml
new file mode 100644
index 0000000..66e273f
--- /dev/null
+++ b/.github/workflows/cloud-write.yml
@@ -0,0 +1,114 @@
+# Destructive ACS writes against a real Splunk Cloud stack, with undo.
+#
+# This never runs on pull_request or a schedule. A human must dispatch it and
+# type WRITE. HEAD's commit subject must start with "tests: splunk cloud write".
+# Use a non-production stack: the job creates and deletes indexes, roles, and
+# HEC tokens.
+name: Splunk Cloud Write Canary
+
+on:
+  workflow_dispatch:
+    inputs:
+      confirm:
+        description: Type WRITE to run destructive ACS tests on the configured stack
+        required: true
+        type: string
+
+permissions:
+  contents: read
+
+concurrency:
+  group: splunk-cloud-write
+  cancel-in-progress: false
+
+jobs:
+  confirm:
+    name: Confirm WRITE
+    runs-on: ubuntu-latest
+    timeout-minutes: 2
+    steps:
+      - name: Fail closed unless confirm is WRITE
+        env:
+          CONFIRM: ${{ github.event.inputs.confirm }}
+        run: |
+          if [ "$CONFIRM" != "WRITE" ]; then
+            echo "confirm must be exactly WRITE"
+            exit 1
+          fi
+
+  cloud-write:
+    name: Cloud / ACS writes
+    needs: confirm
+    runs-on: ubuntu-latest
+    timeout-minutes: 15
+    steps:
+      - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7
+        with:
+          persist-credentials: false
+
+      - name: Require the write-canary commit prefix
+        run: |
+          subject=$(git log -1 --format=%s)
+          case "$subject" in
+            "tests: splunk cloud write"*) ;;
+            *)
+              echo "HEAD subject must start with: tests: splunk cloud write"
+              exit 1
+              ;;
+          esac
+
+      - name: Check for a configured Cloud stack
+        id: stack
+        env:
+          SPLUNK_URL: ${{ secrets.SPLUNK_URL }}
+          SPLUNK_ACS_TOKEN: ${{ secrets.SPLUNK_ACS_WRITE_TOKEN }}
+        run: bash .github/scripts/detect-cloud-stack.sh
+
+      - name: Require write secrets when dispatched
+        if: steps.stack.outputs.ready != 'true'
+        run: |
+          echo "SPLUNK_URL and SPLUNK_ACS_WRITE_TOKEN are required for a WRITE run"
+          exit 1
+
+      - if: steps.stack.outputs.ready == 'true'
+        uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97 # v7
+        with:
+          python-version: "3.14"
+          cache: pip
+
+      - name: Install project
+        if: steps.stack.outputs.ready == 'true'
+        run: |
+          python -m venv .venv
+          .venv/bin/python -m pip install --require-hashes -r requirements-ci.txt
+          .venv/bin/python -m pip install -e . --no-deps
+
+      - name: Cloud writes (index, role, hec-token) with undo
+        if: steps.stack.outputs.ready == 'true'
+        env:
+          SPLUNK_ACS_LIVE_TEST: "true"
+          SPLUNK_CLOUD_WRITE: "true"
+          SPLUNK_URL: ${{ secrets.SPLUNK_URL }}
+          SPLUNK_ACS_WRITE_TOKEN: ${{ secrets.SPLUNK_ACS_WRITE_TOKEN }}
+          SPLUNK_ACS_TOKEN: ${{ secrets.SPLUNK_ACS_WRITE_TOKEN }}
+          SPLUNK_ACS_STACK: ${{ secrets.SPLUNK_ACS_STACK }}
+          SPLUNK_ACS_BASE_URL: ${{ secrets.SPLUNK_ACS_BASE_URL }}
+          VCT_SPLUNK_REDACT_TARGET: "1"
+          VCT_SPLUNK_AUDIT: ${{ runner.temp }}/vct-splunk-audit.log
+        run: |
+          set -o pipefail
+          set +e
+          .venv/bin/pytest tests/integration/cloud/write \
+          -m "integration and cloud and write" \
+          -q --tb=line -r N \
+          -o addopts='--strict-markers --import-mode=importlib' \
+          --junitxml=cloud-write.xml \
+          | tee cloud-write.log
+          pytest_status=${PIPESTATUS[0]}
+          set -e
+          scan_files=(cloud-write.log cloud-write.xml)
+          if [ -n "${GITHUB_STEP_SUMMARY:-}" ]; then
+            scan_files+=("$GITHUB_STEP_SUMMARY")
+          fi
+          python .github/scripts/scan-cloud-ci-leaks.py "${scan_files[@]}"
+          exit "$pytest_status"
diff --git a/AGENTS.md b/AGENTS.md
index fad0470..f3c995a 100644
--- a/AGENTS.md
+++ b/AGENTS.md
@@ -69,8 +69,11 @@ Two cross-cutting ideas to know about:
   require an explicit app and never silently default to `search`.
 - **Transparent backend.** When `SPLUNK_URL` points at `*.splunkcloud.com`, a
   few reads (`index list`, `role list`, `hec-token list`) route through the
-  Cloud ACS API and writes are refused; everything else talks to splunkd REST.
-  The backend is deduced from the URL — there is no flag to pick it. `splunk
+  Cloud ACS API; everything else talks to splunkd REST. Cloud writes are
+  opt-in and narrow: `SPLUNK_CLOUD_WRITE=true` unlocks `create`/`update`/
+  `delete` for `index`, `role`, and `hec-token` only (never a CLI flag, and
+  enable/disable plus every other resource stay refused regardless). The
+  backend is deduced from the URL — there is no flag to pick it. `splunk
   inspect` reports what the deduced backend supports, offline.
 
 ## Conventions
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 1a814e5..641a143 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -12,9 +12,21 @@ project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
   config get FILE STANZA`. The commands use the normal read namespace, accept a
   file name with or without `.conf`, paginate collection results, and redact
   secret-valued properties.
+- Opt-in Splunk Cloud writes for `index`, `role`, and `hec-token`
+  create/update/delete via ACS. Default remains refuse-before-network; set
+  `SPLUNK_CLOUD_WRITE=true` (no CLI flag) and still pass `--yes` or confirm on
+  a TTY. `--dry-run` sends nothing. A separate `SPLUNK_ACS_WRITE_TOKEN` can
+  scope the write credential apart from the read token.
+- GitHub Actions canaries for live Cloud reads (existing workflow, leak-scanned)
+  and destructive Cloud writes with undo (`workflow_dispatch` only, typed
+  `confirm=WRITE`).
 
 ### Changed
 
+- `splunk inspect` no longer emits the Cloud stack name. It reports
+  `stack_configured` instead. Set `VCT_SPLUNK_REDACT_TARGET=1` to hide the
+  stack label in `meta.target` as well (CI does this). The audit log still
+  records a credential-stripped but host-honest target.
 - Lower the supported Python floor to 3.9, so the CLI runs under the interpreter
   bundled with Splunk Enterprise 9.x. Shipped code needed no change: the package
   already uses only 3.9-compatible syntax and APIs. Declarations move
diff --git a/README.md b/README.md
index 4ec105d..3ac26ae 100644
--- a/README.md
+++ b/README.md
@@ -178,10 +178,19 @@ export SPLUNK_ACS_TOKEN="<your ACS token>"
 export SPLUNK_ACS_BASE_URL="https://admin.splunkcloudgc.com"   # only for FedRAMP
 ```
 
-Cloud support is **read-only** today, and covers `index list`, `role list`, and
-`hec-token list`. Anything else stops with a clear "not supported here" error
-instead of guessing. Run `splunk inspect` to see which backend your address
-resolves to and what it can do; it answers offline, without contacting anything.
+Cloud reads cover `index list`, `role list`, and `hec-token list`. Anything else
+stops with a clear "not supported here" error instead of guessing. Run
+`splunk inspect` to see which backend your address resolves to and what it can
+do; it answers offline, without contacting anything.
+
+Cloud **writes** are opt-in and narrow: set `SPLUNK_CLOUD_WRITE=true` to unlock
+`create`/`update`/`delete` for `index`, `role`, and `hec-token` only (enable and
+disable, and every other resource, stay refused regardless). There is no CLI
+flag for the opt-in -- only the environment variable, so a write is never
+enabled by accident from a saved command line. `--dry-run` and `--yes` work the
+same as they do against Enterprise. An ACS write token can be scoped separately
+from the read token via `SPLUNK_ACS_WRITE_TOKEN` (falls back to
+`SPLUNK_ACS_TOKEN` when unset).
 
 ## Security
 
diff --git a/VERSION b/VERSION
index 0d91a54..1d0ba9e 100644
--- a/VERSION
+++ b/VERSION
@@ -1 +1 @@
-0.3.0
+0.4.0
diff --git a/docs/architecture.md b/docs/architecture.md
index 37c962f..90b9b32 100644
--- a/docs/architecture.md
+++ b/docs/architecture.md
@@ -20,7 +20,8 @@ vct_splunk/
       server.py, search.py, jobs.py, saved_searches.py, kvstore.py,
       hec.py, apps.py, cluster.py, license.py, deploy.py, lookups.py,
       datamodel.py, health.py, raw.py
-    acs/                   # Splunk Cloud ACS management-plane client (read-only)
+    acs/                   # Splunk Cloud ACS management-plane client (reads +
+                           # gated index/role/hec-token writes)
       client.py, operations.py
 
   auth/
@@ -31,7 +32,8 @@ vct_splunk/
     registry.py            # Declarative list of factory-generated resources
     context.py             # Shared `command` decorator + Ctx (builds clients)
     write.py               # The single gated write path (confirm + audit)
-    dispatch.py            # Routes index/role/hec-token list to ACS on Cloud
+    dispatch.py            # Routes index/role/hec-token reads (and gated
+                           # writes) to ACS on Cloud
     server.py, api.py, auth.py, search.py, saved_search.py, health.py,
     kvstore.py, hec.py, apps.py, cluster.py, shcluster.py, license.py,
     deploy.py, lookup.py, datamodel.py, inspect.py
@@ -150,8 +152,9 @@ Every mutation funnels through `commands/write.py:do_write()`: `--dry-run`
 previews the exact request and sends nothing; otherwise it confirms on a TTY or
 requires `--yes` when non-interactive, then appends a record to the local audit
 log. Reads redact secret-named fields by default; only commands whose purpose
-is to mint a credential reveal one. Splunk Cloud targets refuse writes and
-route supported reads through ACS.
+is to mint a credential reveal one. Splunk Cloud targets route supported reads
+through ACS; writes are refused by default and opt in narrowly via
+`SPLUNK_CLOUD_WRITE=true` (index/role/hec-token create/update/delete only).
 
 ### Error handling
 
diff --git a/src/vct_splunk/api/acs/__init__.py b/src/vct_splunk/api/acs/__init__.py
index 3fe49bf..2f3faa4 100644
--- a/src/vct_splunk/api/acs/__init__.py
+++ b/src/vct_splunk/api/acs/__init__.py
@@ -1 +1,6 @@
-"""Read-only Splunk Cloud ACS (adminconfig/v2) client and operations."""
+"""Splunk Cloud ACS (adminconfig/v2) client and operations.
+
+Reads are unrestricted; writes exist for index/role/hec-token create, update,
+and delete but are opt-in and gated -- see
+:func:`vct_splunk.commands.write.refuse_cloud_write`.
+"""
diff --git a/src/vct_splunk/api/acs/client.py b/src/vct_splunk/api/acs/client.py
index 08cf519..219441a 100644
--- a/src/vct_splunk/api/acs/client.py
+++ b/src/vct_splunk/api/acs/client.py
@@ -1,10 +1,14 @@
-"""A thin, read-only client for the Splunk Cloud ACS adminconfig/v2 API.
+"""A thin client for the Splunk Cloud ACS adminconfig/v2 API.
 
 ACS is a different surface from splunkd: a different base URL
 (``https://admin.splunk.com/<stack>/adminconfig/v2``), a stack auth token, and
 plain JSON responses (not the form-encoded ``entry[].content`` shape). So it gets
 its own small client rather than reusing :class:`~vct_splunk.api.client.SplunkClient`.
-Writes are intentionally absent this release.
+
+Reads are always available. Writes exist too, but only through the gated path in
+:mod:`vct_splunk.commands.write` -- this client sends whatever it is told, the
+same way :class:`~vct_splunk.api.client.SplunkClient` does; the opt-in gate and
+resource allowlist live one layer up.
 """
 
 from __future__ import annotations
@@ -18,7 +22,7 @@
 import httpx
 
 from ...utils.errors import APIError, AuthError, NotFoundError, TransportError, UsageError
-from ...utils.redact import safe_target
+from ...utils.redact import public_target, redact_exception_text
 
 ACS_BASE_URL = "https://admin.splunk.com"
 _STACK_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9-]*$")
@@ -31,25 +35,35 @@ class AcsConfig:
     token: str
     base_url: str = ACS_BASE_URL
     timeout: float = 30.0
+    dry_run: bool = False
 
 
-def acs_config_from_env(stack: str | None = None) -> AcsConfig:
-    """Build an ACS config: the stack (derived from SPLUNK_URL) + ``SPLUNK_ACS_TOKEN``.
+def acs_config_from_env(stack: str | None = None, *, write: bool = False) -> AcsConfig:
+    """Build an ACS config: the stack (derived from SPLUNK_URL) + an ACS token.
 
     The Cloud stack is normally derived from the ``*.splunkcloud.com`` host in
     ``SPLUNK_URL`` and passed in as ``stack``; ``SPLUNK_ACS_STACK`` is a rare
-    explicit override. ``SPLUNK_ACS_TOKEN`` (a Bearer token, separate from the
-    Enterprise auth token) is always required for ACS operations.
+    explicit override.
+
+    Args:
+        stack: The Cloud stack name, or None to require ``SPLUNK_ACS_STACK``.
+        write: When True, prefer ``SPLUNK_ACS_WRITE_TOKEN`` (falling back to
+            ``SPLUNK_ACS_TOKEN`` when unset) -- lets an operator scope a
+            write-capable token separately from the read token. Reads always
+            use ``SPLUNK_ACS_TOKEN`` only, never the write token.
     """
     stack = os.environ.get("SPLUNK_ACS_STACK") or stack
     token = os.environ.get("SPLUNK_ACS_TOKEN")
+    if write:
+        token = os.environ.get("SPLUNK_ACS_WRITE_TOKEN") or token
     if not stack:
         raise UsageError(
             "Could not determine the Splunk Cloud stack. Set SPLUNK_URL to your "
             "https://<stack>.splunkcloud.com host (or set SPLUNK_ACS_STACK)."
         )
     if not token:
-        raise UsageError("No ACS token. Set SPLUNK_ACS_TOKEN for Splunk Cloud operations.")
+        env_name = "SPLUNK_ACS_WRITE_TOKEN or SPLUNK_ACS_TOKEN" if write else "SPLUNK_ACS_TOKEN"
+        raise UsageError(f"No ACS token. Set {env_name} for Splunk Cloud operations.")
     if not _STACK_RE.fullmatch(stack):
         raise UsageError(
             "Invalid ACS stack name. Use only letters, numbers, and hyphens, "
@@ -60,7 +74,15 @@ def acs_config_from_env(stack: str | None = None) -> AcsConfig:
 
 
 class AcsClient:
-    """Read-only GET access to one Splunk Cloud stack's ACS adminconfig/v2 API."""
+    """Access to one Splunk Cloud stack's ACS adminconfig/v2 API.
+
+    Reads (:meth:`get`) are unrestricted. Mutations (:meth:`write`) are dry-run
+    gated the same way :meth:`vct_splunk.api.client.SplunkClient.write` is --
+    ``config.dry_run`` sends nothing and returns a preview instead. This client
+    does not itself decide *which* resources may be written or whether the
+    caller opted in; that gate is :func:`vct_splunk.commands.write.refuse_cloud_write`,
+    one layer up.
+    """
 
     def __init__(self, config: AcsConfig, *, transport: httpx.BaseTransport | None = None) -> None:
         if not _STACK_RE.fullmatch(config.stack):
@@ -81,13 +103,46 @@ def __exit__(self, exc_type: object, exc: object, tb: object) -> None:
 
     def get(self, path: str, params: dict[str, Any] | None = None) -> Any:
         """GET an ACS read endpoint and return the parsed JSON."""
+        return self.request("GET", path, params=params)
+
+    def write(self, method: str, path: str, json_body: Any | None = None) -> Any:
+        """Mutating ACS request. When dry_run is set, sends nothing and returns a preview.
+
+        Mirrors :meth:`vct_splunk.api.client.SplunkClient.write`: the caller
+        (an ACS operation function) is trusted to have already decided this
+        mutation is allowed to run; this method only decides whether to send it.
+        """
+        if self.config.dry_run:
+            base = f"{self.config.base_url}/{self.config.stack}/adminconfig/v2"
+            return {
+                "dry_run": True,
+                "request": {"method": method, "path": "/" + path.lstrip("/"), "body": json_body},
+                "target": public_target(base),
+            }
+        return self.request(method, path, json_body=json_body)
+
+    def request(
+        self,
+        method: str,
+        path: str,
+        *,
+        params: dict[str, Any] | None = None,
+        json_body: Any | None = None,
+    ) -> Any:
+        """Send one ACS request and return the parsed JSON response.
+
+        Shared by every read and (non-dry-run) write: retries 429/5xx honoring
+        ``Retry-After``, and maps status codes to the same typed errors GET has
+        always raised.
+        """
         url = "/" + path.lstrip("/")
         for attempt in range(_MAX_RETRIES + 1):
             try:
-                resp = self._http.get(url, params=params)
+                resp = self._http.request(method, url, params=params, json=json_body)
             except httpx.HTTPError as exc:
                 raise TransportError(
-                    f"Could not reach ACS at {safe_target(self.config.base_url)}: {exc}"
+                    f"Could not reach ACS at {public_target(self.config.base_url)}: "
+                    f"{redact_exception_text(str(exc))}"
                 ) from exc
             if (resp.status_code == 429 or 500 <= resp.status_code < 600) and (
                 attempt < _MAX_RETRIES
@@ -99,13 +154,13 @@ def get(self, path: str, params: dict[str, Any] | None = None) -> Any:
             if resp.status_code == 404:
                 raise NotFoundError(f"ACS endpoint not found: {url}")
             if resp.status_code >= 400:
-                raise APIError(f"ACS returned {resp.status_code} for GET {url}")
+                raise APIError(f"ACS returned {resp.status_code} for {method} {url}")
             if not resp.content:
                 return {}
             try:
                 return resp.json()
             except ValueError as exc:
-                raise APIError(f"ACS returned malformed JSON for GET {url}") from exc
+                raise APIError(f"ACS returned malformed JSON for {method} {url}") from exc
         raise TransportError("ACS retries exhausted")  # pragma: no cover
 
 
diff --git a/src/vct_splunk/api/acs/operations.py b/src/vct_splunk/api/acs/operations.py
index c263617..85a2068 100644
--- a/src/vct_splunk/api/acs/operations.py
+++ b/src/vct_splunk/api/acs/operations.py
@@ -1,10 +1,21 @@
-"""Read-only ACS operations."""
+"""ACS operations: unrestricted reads plus the three writable resources.
+
+Only index, role, and HTTP Event Collector token support create/update/delete
+on Splunk Cloud (see ``WRITE_PATHS``, checked against Splunk's public OpenAPI by
+``tests/integration/test_acs_public_spec.py``). Every write goes through
+:meth:`~vct_splunk.api.acs.client.AcsClient.write`, so ``--dry-run`` sends
+nothing here exactly as it does for the Enterprise REST path; the opt-in gate
+and the allowlist of which (resource, verb) pairs even reach these functions
+live one layer up, in :mod:`vct_splunk.commands.write` and
+:mod:`vct_splunk.commands.dispatch`.
+"""
 
 from __future__ import annotations
 
 from typing import Any
 
 from ...utils.errors import APIError
+from ...utils.path import path_segment
 from ...utils.redact import redact_secrets
 from .client import AcsClient
 
@@ -23,6 +34,24 @@
 #: Every ACS path the CLI reads.
 READ_PATHS = tuple(LIST_ENVELOPES)
 
+#: (path template, HTTP method) pairs ACS exposes for create/update/delete on
+#: the three writable resources. Templates use Splunk's own OpenAPI
+#: path-parameter names, so the public-spec drift check
+#: (``tests/integration/test_acs_public_spec.py``) can look each one up
+#: directly against the published contract; the functions below build the real
+#: request path with an actual name instead of the placeholder.
+WRITE_PATHS: tuple[tuple[str, str], ...] = (
+    (INDEXES, "post"),
+    (f"{INDEXES}/{{index}}", "patch"),
+    (f"{INDEXES}/{{index}}", "delete"),
+    (ROLES, "post"),
+    (f"{ROLES}/{{roleName}}", "patch"),
+    (f"{ROLES}/{{roleName}}", "delete"),
+    (HEC_TOKENS, "post"),
+    (f"{HEC_TOKENS}/{{hec}}", "patch"),
+    (f"{HEC_TOKENS}/{{hec}}", "delete"),
+)
+
 
 def list_cloud_indexes(client: AcsClient) -> list[dict[str, Any]]:
     """List indexes on the Cloud stack (ACS)."""
@@ -59,3 +88,72 @@ def _list(client: AcsClient, path: str, envelope: str) -> list[dict[str, Any]]:
         if len(page) < 100:
             return output
         offset += len(page)
+
+
+def create_cloud_index(client: AcsClient, name: str, body: dict[str, Any]) -> Any:
+    """Create an index on the Cloud stack (ACS). Splunk provisions it asynchronously."""
+    return _write(client, "POST", INDEXES, {**body, "name": name})
+
+
+def update_cloud_index(client: AcsClient, name: str, body: dict[str, Any]) -> Any:
+    """Update an index's settings on the Cloud stack (ACS)."""
+    return _write(client, "PATCH", f"{INDEXES}/{_encoded(name)}", body)
+
+
+def delete_cloud_index(client: AcsClient, name: str, body: dict[str, Any] | None = None) -> Any:
+    """Delete an index on the Cloud stack (ACS). Deletion completes asynchronously."""
+    return _write(client, "DELETE", f"{INDEXES}/{_encoded(name)}", body or {})
+
+
+def create_cloud_role(client: AcsClient, name: str, body: dict[str, Any]) -> Any:
+    """Create a role on the Cloud stack (ACS)."""
+    return _write(client, "POST", ROLES, {**body, "name": name})
+
+
+def update_cloud_role(client: AcsClient, name: str, body: dict[str, Any]) -> Any:
+    """Update a role's settings on the Cloud stack (ACS)."""
+    return _write(client, "PATCH", f"{ROLES}/{_encoded(name)}", body)
+
+
+def delete_cloud_role(client: AcsClient, name: str, body: dict[str, Any] | None = None) -> Any:
+    """Delete a role on the Cloud stack (ACS)."""
+    return _write(client, "DELETE", f"{ROLES}/{_encoded(name)}", body or {})
+
+
+def create_hec_token(client: AcsClient, name: str, body: dict[str, Any]) -> Any:
+    """Create an HTTP Event Collector token on the Cloud stack (ACS).
+
+    Unlike the Enterprise ``hec-token create`` (whose response is allowed to
+    reveal the token because it is the only way to learn it), the ACS response
+    is redacted here like every other ACS write. Splunk Cloud CI output must
+    never carry a live credential, so there is no reveal-once escape hatch on
+    this path.
+    """
+    return _write(client, "POST", HEC_TOKENS, {**body, "name": name})
+
+
+def update_hec_token(client: AcsClient, name: str, body: dict[str, Any]) -> Any:
+    """Update an HTTP Event Collector token's settings on the Cloud stack (ACS)."""
+    return _write(client, "PATCH", f"{HEC_TOKENS}/{_encoded(name)}", body)
+
+
+def delete_hec_token(client: AcsClient, name: str, body: dict[str, Any] | None = None) -> Any:
+    """Delete an HTTP Event Collector token on the Cloud stack (ACS)."""
+    return _write(client, "DELETE", f"{HEC_TOKENS}/{_encoded(name)}", body or {})
+
+
+def _encoded(name: str) -> str:
+    """Validate and percent-encode a name for use in an ACS item path."""
+    return path_segment(name, label="name")
+
+
+def _write(client: AcsClient, method: str, path: str, body: dict[str, Any]) -> Any:
+    """Send one ACS mutation and redact any secret before it reaches the caller.
+
+    A dry-run preview (``{"dry_run": True, ...}``) carries no live data and
+    passes through unredacted, same as :meth:`~vct_splunk.api.client.SplunkClient.write`.
+    """
+    result = client.write(method, path, body)
+    if isinstance(result, dict) and result.get("dry_run"):
+        return result
+    return redact_secrets(result)
diff --git a/src/vct_splunk/commands/command_factory.py b/src/vct_splunk/commands/command_factory.py
index f5d596e..509b5a2 100644
--- a/src/vct_splunk/commands/command_factory.py
+++ b/src/vct_splunk/commands/command_factory.py
@@ -23,11 +23,20 @@
 from ..utils.namespace import resolve_ns
 from ..utils.validation import parse_key_value_pairs
 from .context import AliasedGroup, command
-from .dispatch import dispatch_list, has_cloud_list
+from .dispatch import dispatch_list, dispatch_write, has_cloud_list
 from .write import do_write, refuse_cloud_write
 
 _VERB_ALIASES = {"add": "create", "edit": "update", "remove": "delete"}
 
+#: For a Cloud-writable spec, the one typed `Field` (if any) that maps onto an
+#: ACS JSON key. Every other typed field has no ACS equivalent -- `--set` with
+#: ACS's own field name (e.g. `searchableDays`, not a Splunk REST form field)
+#: is the escape hatch for those, same as it is for anything --field-options
+#: does not cover on Enterprise.
+_ACS_FIELD_KEYS: dict[str, dict[str, str]] = {
+    "index": {"max_gb": "maxDataSizeMB"},
+}
+
 
 def _help_for(spec: EndpointConfig, verb: str) -> str:
     """One-line help for a generated command, in the hand-written commands' style."""
@@ -47,6 +56,35 @@ def _help_for(spec: EndpointConfig, verb: str) -> str:
     return texts[verb]
 
 
+def _acs_body(spec: EndpointConfig, fields: dict[str, Any], sets: dict[str, str]) -> dict[str, Any]:
+    """Build the ACS JSON body for a Cloud-routed create/update.
+
+    `--set KEY=VALUE` pairs pass straight through as given -- `KEY` must be the
+    ACS field's own JSON name, which is not always the same as the Splunk REST
+    form field the same option sends on Enterprise. A typed field option (e.g.
+    `--max-gb`) is honored only when this spec declares an ACS-equivalent key
+    in `_ACS_FIELD_KEYS`; anything else raises rather than silently dropping or
+    mis-mapping a value the caller explicitly asked to send.
+    """
+    mapping = _ACS_FIELD_KEYS.get(spec.name, {})
+    by_opt = {f.opt: f for f in spec.fields}
+    body: dict[str, Any] = {}
+    for opt, value in fields.items():
+        if value is None or value == ():
+            continue
+        key = mapping.get(opt)
+        if key is None:
+            dashed = opt.replace("_", "-")
+            raise UsageError(
+                f"--{dashed} has no Splunk Cloud (ACS) equivalent. "
+                f"Use --set with the ACS field's own JSON name instead."
+            )
+        f = by_opt[opt]
+        body[key] = int(float(value) * f.scale) if f.scale else value
+    body.update(sets)
+    return body
+
+
 def _gate_args(
     spec: EndpointConfig, verb: str, name: str, owner, app
 ) -> tuple[str, dict[str, Any]]:
@@ -117,7 +155,15 @@ def _create(ctx, name, **opts) -> None:
                 ctx,
                 action=action,
                 audit_event=event,
-                run=lambda c: res.create(c, name, fields=fields, sets=sets, owner=owner, app=app),
+                run=lambda c: dispatch_write(
+                    ctx,
+                    spec.name,
+                    "create",
+                    name,
+                    c,
+                    lambda rc: res.create(rc, name, fields=fields, sets=sets, owner=owner, app=app),
+                    body=_acs_body(spec, fields, sets) if ctx.backend == "cloud" else None,
+                ),
             )
             out.emit(result, ctx.output_mode, ctx.meta())
 
@@ -138,7 +184,15 @@ def _update(ctx, name, **opts) -> None:
                 ctx,
                 action=action,
                 audit_event=event,
-                run=lambda c: res.update(c, name, fields=fields, sets=sets, owner=owner, app=app),
+                run=lambda c: dispatch_write(
+                    ctx,
+                    spec.name,
+                    "update",
+                    name,
+                    c,
+                    lambda rc: res.update(rc, name, fields=fields, sets=sets, owner=owner, app=app),
+                    body=_acs_body(spec, fields, sets) if ctx.backend == "cloud" else None,
+                ),
             )
             out.emit(result, ctx.output_mode, ctx.meta())
 
@@ -155,7 +209,14 @@ def _delete(ctx, name) -> None:
                 ctx,
                 action=action,
                 audit_event=event,
-                run=lambda c: res.delete(c, name, owner=owner, app=app),
+                run=lambda c: dispatch_write(
+                    ctx,
+                    spec.name,
+                    "delete",
+                    name,
+                    c,
+                    lambda rc: res.delete(rc, name, owner=owner, app=app),
+                ),
             )
             out.emit(result, ctx.output_mode, ctx.meta())
 
diff --git a/src/vct_splunk/commands/context.py b/src/vct_splunk/commands/context.py
index ad827dd..73e46fd 100644
--- a/src/vct_splunk/commands/context.py
+++ b/src/vct_splunk/commands/context.py
@@ -31,7 +31,7 @@
 from ..output import formatter as out
 from ..utils.backends import deduce_backend
 from ..utils.errors import SplunkError
-from ..utils.redact import safe_target
+from ..utils.redact import public_target
 
 if TYPE_CHECKING:
     from ..api.acs.client import AcsClient
@@ -126,7 +126,7 @@ def meta(self) -> dict[str, str | None]:
         Right now this is just the target Splunk URL, so a piece of output can be
         traced back to the instance it came from.
         """
-        return {"target": safe_target(self.base_url or "")}
+        return {"target": public_target(self.base_url or "")}
 
 
 def command(fn: Callable) -> Callable:
diff --git a/src/vct_splunk/commands/dispatch.py b/src/vct_splunk/commands/dispatch.py
index 2f82198..656bd51 100644
--- a/src/vct_splunk/commands/dispatch.py
+++ b/src/vct_splunk/commands/dispatch.py
@@ -6,6 +6,11 @@
 with a clean :class:`UnsupportedBackendError` rather than falling through to an
 unofficial endpoint. This is the one place that knows both clients exist; the
 Click-free core stays unaware of backends.
+
+:func:`dispatch_write` is the write-side counterpart, used from inside
+:func:`vct_splunk.commands.write.do_write`'s gated ``run`` callback. Cloud writes
+are opt-in and narrow (see that module for the gate); this module only knows
+*which* (resource, verb) pairs have an ACS route at all.
 """
 
 from __future__ import annotations
@@ -24,6 +29,21 @@
     "hec-token": acs.list_hec_tokens,
 }
 
+#: (resource, verb) -> the ACS op, called as ``op(client, name, body)``. Only
+#: create/update/delete for these three resources have a Cloud write route;
+#: every other resource, and enable/disable on these same three, has none.
+_ACS_WRITE: dict[tuple[str, str], Callable[[Any, str, dict[str, Any]], Any]] = {
+    ("index", "create"): acs.create_cloud_index,
+    ("index", "update"): acs.update_cloud_index,
+    ("index", "delete"): acs.delete_cloud_index,
+    ("role", "create"): acs.create_cloud_role,
+    ("role", "update"): acs.update_cloud_role,
+    ("role", "delete"): acs.delete_cloud_role,
+    ("hec-token", "create"): acs.create_hec_token,
+    ("hec-token", "update"): acs.update_hec_token,
+    ("hec-token", "delete"): acs.delete_hec_token,
+}
+
 
 def has_cloud_list(resource: str) -> bool:
     """True if ``resource``'s list is served by ACS on the Cloud backend."""
@@ -45,3 +65,40 @@ def dispatch_list(ctx: Any, resource: str, rest_call: Callable[[Any], Any]) -> A
             return op(c)
     with ctx.client() as c:
         return rest_call(c)
+
+
+def has_cloud_write(resource: str, verb: str | None = None) -> bool:
+    """True if ``(resource, verb)`` is a Cloud-writable mutation via ACS.
+
+    With ``verb`` omitted, true if ``resource`` has any Cloud write route at
+    all (used where only the resource is known yet, e.g. before a verb-specific
+    check runs).
+    """
+    if verb is None:
+        return any(r == resource for r, _ in _ACS_WRITE)
+    return (resource, verb) in _ACS_WRITE
+
+
+def dispatch_write(
+    ctx: Any,
+    resource: str,
+    verb: str,
+    name: str,
+    client: Any,
+    rest_call: Callable[[Any], Any],
+    *,
+    body: dict[str, Any] | None = None,
+) -> Any:
+    """Run one write against the deduced backend: ACS on Cloud, REST otherwise.
+
+    Called from inside :func:`vct_splunk.commands.write.do_write`'s gated
+    ``run`` callback, with the client it already opened for this backend (an
+    ``AcsClient`` on Cloud, a ``SplunkClient`` otherwise) -- no client is opened
+    here. ``do_write`` has already refused a Cloud write whose ``(resource,
+    verb)`` has no ACS route (see ``refuse_cloud_write``), so a lookup miss here
+    would mean that gate was bypassed.
+    """
+    if ctx.backend == "cloud":
+        op = _ACS_WRITE[(resource, verb)]
+        return op(client, name, body or {})
+    return rest_call(client)
diff --git a/src/vct_splunk/commands/inspect.py b/src/vct_splunk/commands/inspect.py
index 2e9c829..eae7056 100644
--- a/src/vct_splunk/commands/inspect.py
+++ b/src/vct_splunk/commands/inspect.py
@@ -6,6 +6,7 @@
 
 from ..output import formatter as out
 from ..utils.backends import inspect_report
+from ..utils.redact import redact_exception_text
 from .context import command
 
 
@@ -20,4 +21,7 @@ def inspect(ctx) -> None:
     capability map: offline, no live instance touched. Unsupported operations are
     named, so a caller never falls through to an unofficial endpoint.
     """
-    out.emit(inspect_report(ctx.base_url), ctx.output_mode, ctx.meta())
+    meta = ctx.meta()
+    if ctx.backend == "cloud":
+        meta["target"] = redact_exception_text(meta["target"] or "")
+    out.emit(inspect_report(ctx.base_url), ctx.output_mode, meta)
diff --git a/src/vct_splunk/commands/write.py b/src/vct_splunk/commands/write.py
index 44f9a33..cb72cfa 100644
--- a/src/vct_splunk/commands/write.py
+++ b/src/vct_splunk/commands/write.py
@@ -12,15 +12,21 @@
 
 from __future__ import annotations
 
+import os
 from collections.abc import Callable
 from typing import Any
 
-from ..api.client import SplunkClient
-from ..config.loader import load_config
+from ..api.acs.client import AcsClient, acs_config_from_env
+from ..config.loader import _resolve_target, load_config
 from ..output import formatter as out
 from ..utils import audit
+from ..utils.backends import cloud_stack_from_url
 from ..utils.errors import UnsupportedBackendError
-from ..utils.redact import safe_target
+from ..utils.redact import public_target, safe_target
+from .dispatch import has_cloud_write
+
+#: Opt-in for Splunk Cloud writes. Never a CLI flag -- see `refuse_cloud_write`.
+CLOUD_WRITE_ENV = "SPLUNK_CLOUD_WRITE"
 
 
 def do_write(
@@ -28,12 +34,12 @@ def do_write(
     *,
     action: str,
     audit_event: dict[str, Any],
-    run: Callable[[SplunkClient], dict[str, Any]],
+    run: Callable[[Any], dict[str, Any]],
     target: str | None = None,
 ) -> dict[str, Any]:
     """Run one gated mutation: confirm it, execute it, and audit it.
 
-    Resolves the target up front (so a missing ``SPLUNK_URL`` / ``SPLUNK_TOKEN``
+    Resolves the target up front (so a missing ``SPLUNK_URL`` / credential
     fails before we prompt), gates the write via :func:`output.confirm_write`
     (dry-run / ``--yes`` / non-interactive fail-fast), runs ``run(client)``, and
     appends an audit record for any real (non-dry-run) write.
@@ -42,7 +48,8 @@ def do_write(
         ctx: The command :class:`~vct_splunk.commands.context.Ctx`.
         action: A human phrase for the prompt, e.g. ``"create index 'web'"``.
         audit_event: Fields to record on a real write; ``target`` is added here.
-        run: Callable given an open client, returning the operation result.
+        run: Callable given an open client (a ``SplunkClient`` on Enterprise, an
+            ``AcsClient`` for an opted-in Cloud write), returning the result.
         target: The Splunk URL; resolved from the environment when omitted.
 
     Returns:
@@ -51,25 +58,51 @@ def do_write(
     # audit_event["action"] is "<resource>.<verb>", e.g. "index.create".
     resource, _, verb = str(audit_event.get("action", "")).partition(".")
     refuse_cloud_write(ctx, resource, verb)
-    target = safe_target(
-        target or load_config(ctx.base_url, profile=getattr(ctx, "profile", None)).base_url
-    )
-    out.confirm_write(ctx, action, target)
-    with ctx.client() as c:
-        result = run(c)
+    backend = getattr(ctx, "backend", "enterprise")
+    if target is None:
+        if backend == "cloud":
+            # A Cloud write authenticates to ACS, not splunkd, so resolving the
+            # target must not go through `load_config` -- that function demands
+            # a splunkd credential (SPLUNK_TOKEN et al.) this path never needs.
+            target = _resolve_target(ctx.base_url, getattr(ctx, "profile", None))[0].base_url
+        else:
+            target = load_config(ctx.base_url, profile=getattr(ctx, "profile", None)).base_url
+    out.confirm_write(ctx, action, public_target(target))
+    if backend == "cloud":
+        stack = cloud_stack_from_url(ctx.base_url)
+        config = acs_config_from_env(stack, write=True)
+        config.dry_run = ctx.dry_run
+        with AcsClient(config) as c:
+            result = run(c)
+    else:
+        with ctx.client() as c:
+            result = run(c)
     if not (isinstance(result, dict) and result.get("dry_run")):
-        audit.record({**audit_event, "target": target})
+        audit.record({**audit_event, "target": safe_target(target)})
     return result
 
 
 def refuse_cloud_write(ctx: Any, resource: str, verb: str) -> None:
-    """Stop a mutation aimed at a Splunk Cloud stack, before anything else runs.
+    """Stop a mutation aimed at a Splunk Cloud stack, unless explicitly opted in.
+
+    Cloud writes are opt-in and narrow: only create/update/delete for index,
+    role, and hec-token (the ACS mutation routes declared in
+    :mod:`vct_splunk.commands.dispatch`) are ever allowed, and only when
+    ``SPLUNK_CLOUD_WRITE=true`` is set in the environment -- there is no CLI
+    flag for it, so a script cannot flip it on by accident. Every other Cloud
+    mutation -- including enable/disable on those same three resources, and
+    every verb on every other resource -- always raises before any network is
+    touched, opt-in or not. The env var alone never bypasses confirmation:
+    :func:`do_write` still runs `output.confirm_write` (``--yes``/TTY) after
+    this check passes.
 
-    Writes are Enterprise-only this release. :func:`do_write` calls this, which
-    is enough for a command that reaches the gate directly. A command that first
-    resolves a namespace calls it earlier as well, so a Cloud target is told that
-    writes are unsupported rather than being asked for an ``--app`` that would
-    not have helped.
+    :func:`do_write` calls this, which is enough for a command that reaches the
+    gate directly. A command that first resolves a namespace calls it earlier
+    as well, so a Cloud target is told writes are unsupported rather than being
+    asked for an ``--app`` that would not have helped.
     """
-    if getattr(ctx, "backend", "enterprise") == "cloud":
-        raise UnsupportedBackendError(resource or "this resource", verb or "write", "cloud")
+    if getattr(ctx, "backend", "enterprise") != "cloud":
+        return
+    if os.environ.get(CLOUD_WRITE_ENV) == "true" and has_cloud_write(resource, verb):
+        return
+    raise UnsupportedBackendError(resource or "this resource", verb or "write", "cloud")
diff --git a/src/vct_splunk/utils/backends.py b/src/vct_splunk/utils/backends.py
index d896791..d92ee5c 100644
--- a/src/vct_splunk/utils/backends.py
+++ b/src/vct_splunk/utils/backends.py
@@ -19,8 +19,10 @@
 _CLOUD_HOST_MARKER = "splunkcloud"
 
 #: What each backend supports, for the `splunk inspect` report. Values are True
-#: (full support) or a short string naming the limit. Cloud is read-only this
-#: release. This is informational only -- routing is decided per command, and an
+#: (full support) or a short string naming the limit. Cloud writes are opt-in
+#: (`SPLUNK_CLOUD_WRITE=true`) and narrow: only index/role/hec-token create,
+#: update, and delete go through ACS -- everything else on Cloud is read-only.
+#: This is informational only -- routing is decided per command, and an
 #: unavailable operation stops with a typed error, never a silent fallthrough.
 CAPABILITIES: dict[str, dict[str, Any]] = {
     "enterprise": {
@@ -35,11 +37,13 @@
         "health": True,
     },
     "cloud": {
-        "indexes": "read-only (ACS)",
-        "hec_tokens": "read-only (ACS)",
-        "roles": "read-only (ACS)",
+        "indexes": "read via ACS; create/update/delete gated behind SPLUNK_CLOUD_WRITE=true",
+        "hec_tokens": "read via ACS; create/update/delete gated behind SPLUNK_CLOUD_WRITE=true",
+        "roles": "read via ACS; create/update/delete gated behind SPLUNK_CLOUD_WRITE=true",
         "search": "via the search head REST (your SPLUNK_URL), where the stack permits",
-        "writes": "not supported this release (read-only)",
+        "writes": (
+            "opt-in (SPLUNK_CLOUD_WRITE=true): index/role/hec-token create/update/delete only"
+        ),
     },
 }
 
@@ -85,8 +89,10 @@ def inspect_report(url: str | None = None) -> dict[str, Any]:
     backend = deduce_backend(url)
     report: dict[str, Any] = {"backend": backend, "capabilities": CAPABILITIES[backend]}
     if backend == "cloud":
-        report["stack"] = cloud_stack_from_url(url)
+        report["stack_configured"] = cloud_stack_from_url(url) is not None
         report["note"] = (
-            "Cloud/ACS coverage is read-only and not yet certified against a live stack."
+            "Cloud/ACS reads are certified; writes are opt-in (SPLUNK_CLOUD_WRITE=true, "
+            "index/role/hec-token create/update/delete only) and not yet certified against "
+            "a live stack."
         )
     return report
diff --git a/src/vct_splunk/utils/redact.py b/src/vct_splunk/utils/redact.py
index 631445f..b631883 100644
--- a/src/vct_splunk/utils/redact.py
+++ b/src/vct_splunk/utils/redact.py
@@ -14,14 +14,19 @@
 
 from __future__ import annotations
 
+import os
+import re
 from typing import Any
 from urllib.parse import urlsplit, urlunsplit
 
 #: What a hidden value is replaced with. The key stays, so a caller can still
 #: see that the field exists.
 REDACTED = "<redacted>"
+REDACT_TARGET_ENV = "VCT_SPLUNK_REDACT_TARGET"
 
 _SECRET_MARKERS = ("pass4symmkey", "password", "passwd", "secret", "token")
+_CLOUD_STACK_HOST_RE = re.compile(r"(?i)(?<![A-Za-z0-9-])[A-Za-z0-9-]+(?=\.splunkcloud)")
+_ACS_STACK_PATH_RE = re.compile(r"(?i)(admin\.splunk\.com/)(?!<redacted>(?:/|$))[^/?#\s]+")
 
 
 def is_secret_key(key: object) -> bool:
@@ -98,3 +103,39 @@ def safe_target(target: str) -> str:
         pass
     path = parsed.path if "@" not in parsed.path else f"/{REDACTED}"
     return urlunsplit((parsed.scheme, host, path, "", ""))
+
+
+def public_target(target: str) -> str:
+    """Return a credential-safe target, optionally hiding its Cloud stack name.
+
+    ``safe_target`` remains suitable for the audit trail, where a Cloud stack
+    identifies the instance. Set :data:`REDACT_TARGET_ENV` to ``"1"`` for
+    untrusted output boundaries such as CI logs.
+    """
+    target = safe_target(target)
+    if os.environ.get(REDACT_TARGET_ENV) != "1":
+        return target
+    try:
+        parsed = urlsplit(target)
+    except ValueError:
+        return redact_exception_text(target)
+    host = parsed.hostname
+    if host and "splunkcloud" in host.casefold():
+        _, separator, suffix = host.partition(".")
+        if separator:
+            host = f"{REDACTED}.{suffix}"
+            if parsed.port is not None:
+                host = f"{host}:{parsed.port}"
+            return urlunsplit((parsed.scheme, host, parsed.path, "", ""))
+    if host and host.casefold() == "admin.splunk.com":
+        path_parts = parsed.path.split("/")
+        if len(path_parts) > 1 and path_parts[1]:
+            path_parts[1] = REDACTED
+        return urlunsplit((parsed.scheme, host, "/".join(path_parts), "", ""))
+    return target
+
+
+def redact_exception_text(text: str) -> str:
+    """Hide Cloud stack labels from a free-form exception message."""
+    text = _CLOUD_STACK_HOST_RE.sub(REDACTED, text)
+    return _ACS_STACK_PATH_RE.sub(rf"\g<1>{REDACTED}", text)
diff --git a/tests/TESTING.md b/tests/TESTING.md
index 451fa64..104b543 100644
--- a/tests/TESTING.md
+++ b/tests/TESTING.md
@@ -1,6 +1,6 @@
 # Running the tests
 
-Six groups. Only the first needs nothing at all — start there.
+Seven groups. Only the first needs nothing at all — start there.
 
 | Group | What it checks | What you must provide | Directory |
 | --- | --- | --- | --- |
@@ -8,6 +8,7 @@ Six groups. Only the first needs nothing at all — start there.
 | Enterprise reads | Every read command against a real server | A reachable Splunk | `tests/integration/enterprise/read/` |
 | Enterprise writes | Every change, then undoes it | A **disposable** Splunk | `tests/integration/enterprise/write/` |
 | Cloud reads | Every read command against a real Cloud stack | A Cloud stack and an ACS token | `tests/integration/cloud/read/` |
+| Cloud writes | Index/role/HEC create-update-delete, then undo | Non-production Cloud stack + write ACS token | `tests/integration/cloud/write/` |
 | ACS contract | Whether Splunk changed its public Cloud API | Nothing | `tests/integration/` |
 | Fuzz | That a credentialed URL never survives redaction | Linux on x86_64 | `tests/fuzz/` |
 
@@ -109,7 +110,36 @@ export SPLUNK_TEST_SERVER_FIXTURE_DIR=/opt/splunk/var/run/splunk/lookup_tmp
 
 Clean up when you are finished: `docker rm -f splunk-test`.
 
-## Group 4: human-operated Splunk Cloud validation
+## Group 4: Splunk Cloud validation (GitHub Actions, then local fallback)
+
+Certification is the GitHub Actions canary, not a filled-in table. Configure
+these repository secrets (names only; never commit values):
+
+- Read canary (`Splunk Cloud Read Canary`): `SPLUNK_URL`, `SPLUNK_ACS_TOKEN`,
+  `SPLUNK_ACS_STACK` (the stack label, so leftover prints still mask), optional
+  `SPLUNK_ACS_BASE_URL`, optional `SPLUNK_TOKEN` for the full catalogue vs
+  ACS-only.
+- Write canary (`Splunk Cloud Write Canary`): `SPLUNK_ACS_WRITE_TOKEN` (a
+  write-scoped ACS JWT, distinct from the read token). Point it at a
+  **non-production** stack. The job creates and deletes indexes, roles, and
+  HEC tokens; that consumes entitlement and appears in the stack's own audit
+  trail.
+
+Dispatch the read workflow from the Actions tab (or `gh workflow run`). A
+100% pass with a clean leak scan is the Cloud read certification. The write
+workflow never runs on a schedule or a pull request: type `WRITE` as the
+`confirm` input, and HEAD's commit subject must start with
+`tests: splunk cloud write`. An optional GitHub Environment
+`splunk-cloud-write` with required reviewers is extra defence; the workflow
+does not require it so a missing Environment cannot block a first run.
+
+Store throwaway fake secret values first, dispatch, confirm the leak scan is
+clean, then swap in real credentials. A leaked ACS token can be rotated; a
+leaked stack name cannot.
+
+The human-operated runbook below remains as a local fallback.
+
+## Group 4b: human-operated Splunk Cloud validation
 
 This is the approval runbook for a real Splunk Cloud stack. It is read-only and
 requires no AI or interpretation service: a person runs the commands, checks
diff --git a/tests/integration/cloud/conftest.py b/tests/integration/cloud/conftest.py
index ee51101..13cb5d8 100644
--- a/tests/integration/cloud/conftest.py
+++ b/tests/integration/cloud/conftest.py
@@ -28,8 +28,7 @@ def _require_cloud_target() -> None:
     # would quietly test Enterprise instead of what this suite claims to cover.
     if deduce_backend(url) != "cloud":
         pytest.fail(
-            f"SPLUNK_URL={url!r} is not a Splunk Cloud host; "
-            "expected something like https://<stack>.splunkcloud.com",
+            "SPLUNK_URL is not a Splunk Cloud host; expected a *.splunkcloud.com URL",
             pytrace=False,
         )
 
diff --git a/tests/integration/cloud/write/conftest.py b/tests/integration/cloud/write/conftest.py
new file mode 100644
index 0000000..de4660a
--- /dev/null
+++ b/tests/integration/cloud/write/conftest.py
@@ -0,0 +1,116 @@
+"""Shared safety gate and CLI harness for destructive Splunk Cloud tests.
+
+Nested under `tests/integration/cloud/`, so `_require_cloud_target` (that
+package's `conftest.py`) already gates `SPLUNK_ACS_LIVE_TEST`, a Cloud
+`SPLUNK_URL`, and `SPLUNK_ACS_TOKEN` before any test here runs. This file adds
+the write-specific opt-in on top, mirroring
+`tests/integration/enterprise/write/conftest.py`.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import uuid
+from collections.abc import Callable, Iterator
+from dataclasses import dataclass, field
+from typing import Any
+
+import pytest
+from click.testing import CliRunner, Result
+
+from vct_splunk.cli import cli
+
+#: Groups every name this test run creates, so a stale object from another run
+#: (or one still mid-async-delete) is never mistaken for one of this run's own.
+RUN_ID = os.environ.get("GITHUB_RUN_ID") or uuid.uuid4().hex[:8]
+
+
+def unique_name(prefix: str) -> str:
+    """A `vct_ci_<run>_<prefix>_<random>` name unique to this test run."""
+    return f"vct_ci_{RUN_ID}_{prefix}_{uuid.uuid4().hex[:8]}"
+
+
+@pytest.fixture(autouse=True)
+def _require_cloud_write_opt_in() -> None:
+    if os.environ.get("SPLUNK_CLOUD_WRITE") == "true":
+        return
+    message = "set SPLUNK_CLOUD_WRITE=true to run destructive Splunk Cloud tests"
+    if os.environ.get("CI") == "true":
+        pytest.fail(message, pytrace=False)
+    pytest.skip(message)
+
+
+@pytest.fixture(scope="session", autouse=True)
+def _preflight_leftover_report() -> None:
+    """List existing `vct_ci_*` objects before this run starts, without failing.
+
+    ACS index and HEC-token deletes complete asynchronously, so an object from
+    a run that finished (and whose own cleanup succeeded) moments ago can still
+    be listed here. Asserting on that would be a false failure, so this is a
+    report only. The enforceable guarantee is narrower and per-test: each test
+    below polls its own delete to completion before returning, and
+    `CloudCli.finish()` fails loudly if any registered reverse-cleanup command
+    itself errors.
+    """
+    runner = CliRunner()
+    for resource in ("index", "role", "hec-token"):
+        result = runner.invoke(cli, [resource, "list", "--output", "json"])
+        if result.exit_code != 0:
+            continue
+        payload = json.loads(result.stdout)
+        leftover = [
+            item.get("name")
+            for item in payload.get("data", [])
+            if isinstance(item, dict) and str(item.get("name", "")).startswith("vct_ci_")
+        ]
+        if leftover:
+            print(f"[cloud write preflight] leftover {resource} objects: {leftover}")
+
+
+@dataclass
+class CloudCli:
+    """Invoke the public CLI against Splunk Cloud and fail if reverse cleanup leaks state."""
+
+    runner: CliRunner = field(default_factory=CliRunner)
+    cleanups: list[tuple[str, Callable[[], Result]]] = field(default_factory=list)
+
+    def run(self, *argv: str, exit_codes: tuple[int, ...] = (0,)) -> Any:
+        result = self.runner.invoke(cli, [*argv, "--output", "json"])
+        assert result.exit_code in exit_codes, (
+            f"{' '.join(argv)} exited {result.exit_code}\n{result.output}"
+        )
+        payload = json.loads(result.stdout)
+        if result.exit_code == 0:
+            assert set(payload) == {"data", "meta"}
+            return payload["data"]
+        return payload["error"]
+
+    def write(self, *argv: str) -> dict[str, Any]:
+        return self.run(*argv, "--yes")
+
+    def cleanup(self, label: str, *argv: str) -> None:
+        self.cleanups.append(
+            (label, lambda: self.runner.invoke(cli, [*argv, "--yes", "--output", "json"]))
+        )
+
+    def drop_cleanup(self, label: str) -> None:
+        """Remove a previously registered cleanup, e.g. once a test's own delete succeeds."""
+        self.cleanups = [c for c in self.cleanups if c[0] != label]
+
+    def finish(self) -> None:
+        failures: list[str] = []
+        for label, cleanup in reversed(self.cleanups):
+            result = cleanup()
+            if result.exit_code != 0:
+                failures.append(f"{label}: exit {result.exit_code}: {result.output}")
+        assert not failures, "cleanup failures:\n" + "\n".join(failures)
+
+
+@pytest.fixture
+def cloud_cli() -> Iterator[CloudCli]:
+    harness = CloudCli()
+    try:
+        yield harness
+    finally:
+        harness.finish()
diff --git a/tests/integration/cloud/write/test_write_catalog.py b/tests/integration/cloud/write/test_write_catalog.py
new file mode 100644
index 0000000..097339b
--- /dev/null
+++ b/tests/integration/cloud/write/test_write_catalog.py
@@ -0,0 +1,86 @@
+"""Apply every Cloud-writable mutation against a real ACS-backed stack, then undo it.
+
+The write-side mirror of `tests/integration/enterprise/write/test_write_catalog.py`,
+narrowed to the three resources ACS actually lets this CLI mutate: `index`,
+`role`, and `hec-token`. No restart, no app install, no file inputs -- ACS has
+no equivalent surface for any of those.
+
+Each test creates one uniquely named object, updates it, deletes it, and polls
+the resource's list until the name is gone (ACS index and HEC-token deletes
+complete asynchronously, so a `202` response does not mean the object is gone
+yet). `cloud_cli.finish()` still fails loudly if the registered reverse
+cleanup itself errors -- the poll only proves the delete this test issued
+actually completed.
+"""
+
+from __future__ import annotations
+
+import time
+
+import pytest
+
+from .conftest import CloudCli, unique_name
+
+pytestmark = [pytest.mark.integration, pytest.mark.cloud, pytest.mark.write]
+
+_POLL_TIMEOUT = 120.0
+_POLL_INTERVAL = 3.0
+
+
+def _poll_until_gone(cloud_cli: CloudCli, resource: str, name: str) -> None:
+    """Poll `resource list` until *name* no longer appears, or fail after the timeout."""
+    deadline = time.monotonic() + _POLL_TIMEOUT
+    while time.monotonic() < deadline:
+        items = cloud_cli.run(resource, "list")
+        if not any(isinstance(item, dict) and item.get("name") == name for item in items):
+            return
+        time.sleep(_POLL_INTERVAL)
+    pytest.fail(f"{resource} {name!r} is still listed {_POLL_TIMEOUT:.0f}s after delete")
+
+
+def test_index_create_update_delete(cloud_cli: CloudCli) -> None:
+    """Create an index, resize it, delete it, and wait for ACS to finish deleting it."""
+    name = unique_name("index")
+    label = f"delete index {name}"
+    cloud_cli.cleanup(label, "index", "delete", name)
+
+    cloud_cli.write("index", "create", name)
+    cloud_cli.write("index", "update", name, "--max-gb", "2")
+    cloud_cli.write("index", "delete", name)
+
+    cloud_cli.drop_cleanup(label)  # already deleted above; nothing left to undo
+    _poll_until_gone(cloud_cli, "index", name)
+
+
+def test_role_create_update_delete(cloud_cli: CloudCli) -> None:
+    """Create a role, change a setting, delete it, and confirm it is gone."""
+    name = unique_name("role")
+    label = f"delete role {name}"
+    cloud_cli.cleanup(label, "role", "delete", name)
+
+    cloud_cli.write("role", "create", name, "--set", "defaultApp=search")
+    cloud_cli.write("role", "update", name, "--set", "srchFilter=search index=main")
+    cloud_cli.write("role", "delete", name)
+
+    cloud_cli.drop_cleanup(label)
+    _poll_until_gone(cloud_cli, "role", name)
+
+
+def test_hec_token_create_update_delete(cloud_cli: CloudCli) -> None:
+    """Create an HEC token, change a setting, delete it, and confirm it is gone.
+
+    Also proves the ACS create response never shows the minted token -- unlike
+    the Enterprise `hec-token create`, which is allowed to reveal it once.
+    """
+    name = unique_name("hec")
+    label = f"delete hec-token {name}"
+    cloud_cli.cleanup(label, "hec-token", "delete", name)
+
+    created = cloud_cli.write("hec-token", "create", name, "--set", "defaultIndex=main")
+    assert created.get("token") in (None, "<redacted>")
+
+    cloud_cli.write("hec-token", "update", name, "--set", "defaultSourcetype=vct_ci")
+    cloud_cli.write("hec-token", "delete", name)
+
+    cloud_cli.drop_cleanup(label)
+    _poll_until_gone(cloud_cli, "hec-token", name)
diff --git a/tests/integration/test_acs_public_spec.py b/tests/integration/test_acs_public_spec.py
index 5a64e88..ee63638 100644
--- a/tests/integration/test_acs_public_spec.py
+++ b/tests/integration/test_acs_public_spec.py
@@ -7,7 +7,7 @@
 import httpx
 import pytest
 
-from vct_splunk.api.acs.operations import LIST_ENVELOPES
+from vct_splunk.api.acs.operations import LIST_ENVELOPES, WRITE_PATHS
 
 pytestmark = [
     pytest.mark.integration,
@@ -31,3 +31,13 @@ def test_implemented_acs_contract_matches_public_spec():
         )
         schema = operation["responses"]["200"]["content"]["application/json"]["schema"]
         assert schema["properties"][envelope]["type"] == "array"
+
+
+def test_implemented_acs_write_contract_matches_public_spec():
+    if os.environ.get("SPLUNK_ACS_SPEC_TEST") != "true":
+        pytest.skip("set SPLUNK_ACS_SPEC_TEST=true to check the public ACS OpenAPI")
+
+    public = httpx.get(_SOURCE, timeout=30).raise_for_status().json()
+    for path, method in WRITE_PATHS:
+        operations = public["paths"][f"/{{stack}}/adminconfig/v2/{path}"]
+        assert method in operations, f"{method.upper()} {path} is missing from the public spec"
diff --git a/tests/unit/test_acs.py b/tests/unit/test_acs.py
index 61a1a47..9d09545 100644
--- a/tests/unit/test_acs.py
+++ b/tests/unit/test_acs.py
@@ -169,6 +169,19 @@ def handler(req: httpx.Request) -> httpx.Response:
         operations.list_cloud_roles(_acs(handler))
 
 
+def test_acs_transport_error_hides_stack_from_exception_text():
+    def handler(req: httpx.Request) -> httpx.Response:
+        raise httpx.ConnectError(
+            "failed to reach https://admin.splunk.com/acme/adminconfig/v2/roles"
+        )
+
+    with pytest.raises(TransportError) as exc_info:
+        operations.list_cloud_roles(_acs(handler))
+
+    assert "acme" not in str(exc_info.value)
+    assert "admin.splunk.com/<redacted>/adminconfig/v2/roles" in str(exc_info.value)
+
+
 def test_acs_config_requires_stack(monkeypatch):
     monkeypatch.delenv("SPLUNK_ACS_STACK", raising=False)
     monkeypatch.delenv("SPLUNK_ACS_TOKEN", raising=False)
@@ -296,7 +309,9 @@ def test_inspect_reports_deduced_cloud(monkeypatch):
     result = CliRunner().invoke(cli, ["inspect", "--output", "json"])
     assert result.exit_code == 0
     assert '"backend": "cloud"' in result.output
-    assert '"stack": "acme"' in result.output
+    assert '"stack_configured": true' in result.output
+    assert '"stack": "acme"' not in result.output
+    assert "acme" not in result.output
     assert "not yet certified" in result.output
 
 
diff --git a/tests/unit/test_cloud_write_opt_in.py b/tests/unit/test_cloud_write_opt_in.py
new file mode 100644
index 0000000..ccf3446
--- /dev/null
+++ b/tests/unit/test_cloud_write_opt_in.py
@@ -0,0 +1,142 @@
+"""Splunk Cloud writes are opt-in and narrow.
+
+``SPLUNK_CLOUD_WRITE=true`` unlocks exactly nine leaves: create/update/delete for
+`index`, `role`, and `hec-token` -- the three ACS mutation routes. Every other
+Cloud mutation, including enable/disable on those same three resources, stays
+refused before any network is touched even with the opt-in set; there is no CLI
+flag for the opt-in, only the environment variable.
+
+This is the counterpart to `test_cloud_write_refusal.py`, which proves the
+*default* (no opt-in) behavior. Neither needs a Cloud stack.
+"""
+
+from __future__ import annotations
+
+import json
+
+import httpx
+import pytest
+from click.testing import CliRunner
+
+from cli_catalog import CATALOG, Case
+from vct_splunk.api.acs import operations as acs
+from vct_splunk.api.acs.client import AcsClient as RealAcsClient
+from vct_splunk.cli import cli
+from vct_splunk.commands.dispatch import has_cloud_write
+from vct_splunk.utils.errors import UnsupportedBackendError
+
+#: CLI resource name -> the ACS collection path it writes to.
+_ACS_PATH = {"index": acs.INDEXES, "role": acs.ROLES, "hec-token": acs.HEC_TOKENS}
+
+WRITE_CASES = tuple(case for case in CATALOG if case.kind == "write")
+CLOUD_WRITABLE_CASES = tuple(
+    case for case in WRITE_CASES if has_cloud_write(case.path[0], case.path[-1])
+)
+STILL_REFUSED_CASES = tuple(
+    case for case in WRITE_CASES if not has_cloud_write(case.path[0], case.path[-1])
+)
+
+
+@pytest.fixture
+def cloud_write_env(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Point the CLI at a Cloud stack with the write opt-in set."""
+    for name in ("SPLUNK_APP", "SPLUNK_OWNER", "SPLUNK_PROFILE", "SPLUNK_ACS_BASE_URL"):
+        monkeypatch.delenv(name, raising=False)
+    monkeypatch.setenv("SPLUNK_URL", "https://acme.splunkcloud.com")
+    monkeypatch.setenv("SPLUNK_ACS_TOKEN", "T")
+    monkeypatch.setenv("SPLUNK_CLOUD_WRITE", "true")
+
+
+@pytest.fixture
+def refuse_network(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Make any real request an error, so a passing test proves nothing left the process."""
+
+    def refuse(*args: object, **kwargs: object) -> object:
+        raise AssertionError("a write reached the network")
+
+    monkeypatch.setattr(httpx.HTTPTransport, "handle_request", refuse)
+
+
+def _patch_acs_write(monkeypatch: pytest.MonkeyPatch, handler) -> None:
+    """Back the ACS write client `do_write` opens with a `httpx.MockTransport`."""
+
+    def _make(config):
+        return RealAcsClient(config, transport=httpx.MockTransport(handler))
+
+    monkeypatch.setattr("vct_splunk.commands.write.AcsClient", _make)
+
+
+def _argv(case: Case, *extra: str) -> list[str]:
+    """Build one invocation of *case*, replacing any `--dry-run` with *extra*."""
+    args = [arg for arg in case.argvs[0] if arg != "--dry-run"]
+    return [*case.path, *args, *extra, "--output", "json"]
+
+
+@pytest.mark.parametrize("case", STILL_REFUSED_CASES, ids=lambda case: " ".join(case.path))
+def test_non_acs_writes_still_refused_when_opted_in(
+    case: Case, cloud_write_env: None, refuse_network: None
+) -> None:
+    """The opt-in unlocks only the nine ACS routes -- everything else still refuses."""
+    result = CliRunner().invoke(cli, _argv(case, "--yes"))
+
+    assert result.exit_code == UnsupportedBackendError.exit_code, (
+        f"{' '.join(case.path)} exited {result.exit_code}: {result.output}"
+    )
+    payload = json.loads(result.output)
+    assert set(payload) == {"error"}
+    assert payload["error"]["code"] == UnsupportedBackendError.code
+
+
+@pytest.mark.parametrize("case", CLOUD_WRITABLE_CASES, ids=lambda case: " ".join(case.path))
+def test_acs_write_dry_run_sends_nothing(
+    case: Case, cloud_write_env: None, refuse_network: None
+) -> None:
+    """--dry-run previews an ACS write and sends no request, even when opted in."""
+    result = CliRunner().invoke(cli, _argv(case, "--dry-run"))
+
+    assert result.exit_code == 0, result.output
+    payload = json.loads(result.output)
+    assert payload["data"]["dry_run"] is True
+    assert payload["data"]["request"]["method"] in {"POST", "PATCH", "DELETE"}
+
+
+@pytest.mark.parametrize("case", CLOUD_WRITABLE_CASES, ids=lambda case: " ".join(case.path))
+def test_acs_write_succeeds_when_opted_in_and_confirmed(
+    case: Case, cloud_write_env: None, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """With the opt-in and --yes, a mocked ACS create/update/delete succeeds."""
+    seen: dict[str, str] = {}
+
+    def handler(req: httpx.Request) -> httpx.Response:
+        seen["method"] = req.method
+        seen["path"] = req.url.path
+        return httpx.Response(202, json={"name": "example"})
+
+    _patch_acs_write(monkeypatch, handler)
+    result = CliRunner().invoke(cli, _argv(case, "--yes"))
+
+    assert result.exit_code == 0, result.output
+    resource, verb = case.path
+    expected_method = {"create": "POST", "update": "PATCH", "delete": "DELETE"}[verb]
+    assert seen["method"] == expected_method
+    assert seen["path"].startswith(f"/acme/adminconfig/v2/{_ACS_PATH[resource]}")
+
+
+def test_hec_token_create_redacts_token_in_output(
+    cloud_write_env: None, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """ACS mints the token on create; Cloud CI output must never show it."""
+
+    def handler(req: httpx.Request) -> httpx.Response:
+        return httpx.Response(
+            202, json={"http-event-collector": {"name": "example", "token": "SECRET-DO-NOT-LEAK"}}
+        )
+
+    _patch_acs_write(monkeypatch, handler)
+    result = CliRunner().invoke(
+        cli, ["hec-token", "create", "example", "--yes", "--output", "json"]
+    )
+
+    assert result.exit_code == 0, result.output
+    assert "SECRET-DO-NOT-LEAK" not in result.output
+    assert "<redacted>" in result.output
diff --git a/tests/unit/test_cloud_write_refusal.py b/tests/unit/test_cloud_write_refusal.py
index f5b4435..8db9db7 100644
--- a/tests/unit/test_cloud_write_refusal.py
+++ b/tests/unit/test_cloud_write_refusal.py
@@ -38,8 +38,16 @@ def cloud_target_with_no_network(monkeypatch: pytest.MonkeyPatch) -> None:
     fails earlier for an unrelated reason.
     """
     # Clear the ambient settings a developer may have exported, so the suite
-    # tests the tool rather than the machine it runs on.
-    for name in ("SPLUNK_APP", "SPLUNK_OWNER", "SPLUNK_PROFILE", "SPLUNK_ACS_BASE_URL"):
+    # tests the tool rather than the machine it runs on. SPLUNK_CLOUD_WRITE in
+    # particular: this suite must prove every write is refused by *default*,
+    # which an ambient opt-in in the environment would silently defeat.
+    for name in (
+        "SPLUNK_APP",
+        "SPLUNK_OWNER",
+        "SPLUNK_PROFILE",
+        "SPLUNK_ACS_BASE_URL",
+        "SPLUNK_CLOUD_WRITE",
+    ):
         monkeypatch.delenv(name, raising=False)
     monkeypatch.setenv("SPLUNK_URL", "https://acme.splunkcloud.com")
     monkeypatch.setenv("SPLUNK_ACS_TOKEN", "unused")
diff --git a/tests/unit/test_public_target.py b/tests/unit/test_public_target.py
new file mode 100644
index 0000000..526647b
--- /dev/null
+++ b/tests/unit/test_public_target.py
@@ -0,0 +1,37 @@
+"""Tests for Cloud stack redaction at user-visible output boundaries."""
+
+from __future__ import annotations
+
+from vct_splunk.utils.redact import (
+    REDACT_TARGET_ENV,
+    public_target,
+    redact_exception_text,
+    safe_target,
+)
+
+
+def test_safe_target_keeps_cloud_hostname_when_public_redaction_is_disabled(monkeypatch):
+    monkeypatch.delenv(REDACT_TARGET_ENV, raising=False)
+
+    assert safe_target("https://acme.splunkcloud.com:8089") == "https://acme.splunkcloud.com:8089"
+
+
+def test_public_target_hides_cloud_stack_when_enabled(monkeypatch):
+    monkeypatch.setenv(REDACT_TARGET_ENV, "1")
+
+    target = public_target("https://acme.splunkcloud.com:8089/services")
+
+    assert "acme" not in target
+    assert target == "https://<redacted>.splunkcloud.com:8089/services"
+
+
+def test_redact_exception_text_hides_acs_and_cloud_stack_labels():
+    text = (
+        "https://admin.splunk.com/acme/adminconfig/v2/indexes and https://acme.splunkcloud.com/foo"
+    )
+
+    redacted = redact_exception_text(text)
+
+    assert "acme" not in redacted
+    assert "https://admin.splunk.com/<redacted>/adminconfig/v2/indexes" in redacted
+    assert "https://<redacted>.splunkcloud.com/foo" in redacted