diff --git a/cdc-service/src/main/java/com/xtrmetl/cdc/replication/SchemaChangeReplicaApplier.java b/cdc-service/src/main/java/com/xtrmetl/cdc/replication/SchemaChangeReplicaApplier.java
index c9f016f8..50228a0a 100644
--- a/cdc-service/src/main/java/com/xtrmetl/cdc/replication/SchemaChangeReplicaApplier.java
+++ b/cdc-service/src/main/java/com/xtrmetl/cdc/replication/SchemaChangeReplicaApplier.java
@@ -24,14 +24,14 @@
  *
  * <p>DDL execution is disabled by default. When enabled, the default policy is a positive
  * allow-list, only a single comment-free statement is accepted, and prefix matches must end
- * at a SQL token boundary.
+ * at a SQL token boundary. Ordinary lifecycle logs retain bounded outcome metadata without
+ * reproducing raw DDL statements or provider exception diagnostics.</p>
  */
 @Service
 @ConditionalOnProperty(prefix = "xtrmetl.replica", name = "enabled", havingValue = "true")
 public class SchemaChangeReplicaApplier {
 
     private static final Logger log = LoggerFactory.getLogger(SchemaChangeReplicaApplier.class);
-    private static final int DDL_LOG_MAX_LENGTH = 500;
     private static final Set<String> IDEMPOTENT_DDL_SQL_STATES = Set.of(
             "42P06",
             "42P07",
@@ -94,20 +94,17 @@ public void apply(@Nullable String topic, @Nullable String keyJson, @Nullable St
             // single-statement, comment-free, and configured policy gates above.
             jdbcTemplate.execute(ddl); // nosemgrep: java.spring.security.audit.spring-sqli.spring-sqli
             if (log.isInfoEnabled()) {
-                log.info("Applied schema change DDL on replica (topic={}, ddl={})",
-                        topic, truncateForLog(ddl));
+                log.info("Applied schema change DDL on replica (topic={})", topic);
             }
         } catch (DataAccessException e) {
             if (isIdempotentDuplicate(e)) {
                 if (log.isInfoEnabled()) {
-                    log.info("Schema change DDL already applied; skipping duplicate (topic={}, ddl={})",
-                            topic, truncateForLog(ddl));
+                    log.info("Schema change DDL already applied; skipping duplicate (topic={})", topic);
                 }
                 return;
             }
             if (log.isErrorEnabled()) {
-                log.error("Failed to apply schema change DDL on replica (topic={}, ddl={})",
-                        topic, truncateForLog(ddl), e);
+                log.error("Failed to apply schema change DDL on replica (topic={})", topic);
             }
             throw e;
         }
@@ -119,7 +116,7 @@ private String requireSingleStatement(String topic, String ddl) {
             trimmed = trimmed.substring(0, trimmed.length() - 1).trim();
         }
         if (trimmed.contains(";")) {
-            logBlocked(topic, ddl, "Blocked multi-statement DDL");
+            logBlocked(topic, "Blocked multi-statement DDL");
             throw new IllegalArgumentException("Multiple SQL statements are not allowed");
         }
         return trimmed;
@@ -127,7 +124,7 @@ private String requireSingleStatement(String topic, String ddl) {
 
     private String requireCommentFree(String topic, String ddl) {
         if (ddl.contains("--") || ddl.contains("/*") || ddl.contains("*/") || ddl.indexOf('\0') >= 0) {
-            logBlocked(topic, ddl, "Blocked DDL containing SQL comments or NUL");
+            logBlocked(topic, "Blocked DDL containing SQL comments or NUL");
             throw new IllegalArgumentException("SQL comments and NUL characters are not allowed in replicated DDL");
         }
         return ddl;
@@ -142,7 +139,7 @@ private String extractDdl(String valueJson) {
             }
             return payload.path("ddl").asText(null);
         } catch (IOException e) {
-            log.warn("Failed to parse Debezium schema change JSON; skipping DDL apply", e);
+            log.warn("Failed to parse Debezium schema change JSON; skipping DDL apply");
             return null;
         }
     }
@@ -230,21 +227,20 @@ private void validateDdl(String topic, String ddl) {
         String normalized = normalizeForValidation(ddl);
         if (ddlValidationMode == DdlValidationMode.BLOCKLIST
                 && ddlBlockedPrefixes.stream().anyMatch(prefix -> matchesPrefix(normalized, prefix))) {
-            logBlocked(topic, ddl, "Blocked DDL by validation policy");
+            logBlocked(topic, "Blocked DDL by validation policy");
             throw new IllegalArgumentException("DDL blocked by validation policy");
         }
 
         if (ddlValidationMode == DdlValidationMode.WHITELIST
                 && ddlAllowedPrefixes.stream().noneMatch(prefix -> matchesPrefix(normalized, prefix))) {
-            logBlocked(topic, ddl, "Blocked DDL by validation policy");
+            logBlocked(topic, "Blocked DDL by validation policy");
             throw new IllegalArgumentException("DDL blocked by validation policy");
         }
     }
 
-    private void logBlocked(String topic, String ddl, String message) {
+    private void logBlocked(String topic, String message) {
         if (log.isWarnEnabled()) {
-            log.warn("{} (mode={}, topic={}, ddl={})",
-                    message, ddlValidationMode, topic, truncateForLog(ddl));
+            log.warn("{} (mode={}, topic={})", message, ddlValidationMode, topic);
         }
     }
 
@@ -268,17 +264,6 @@ private static String normalizeForValidation(String ddl) {
         return ddl.trim().replaceAll("\\s+", " ").toUpperCase(Locale.ROOT);
     }
 
-    private static String truncateForLog(String ddl) {
-        if (ddl == null) {
-            return null;
-        }
-        String normalized = ddl.trim().replaceAll("\\s+", " ");
-        if (normalized.length() <= DDL_LOG_MAX_LENGTH) {
-            return normalized;
-        }
-        return normalized.substring(0, DDL_LOG_MAX_LENGTH) + "...";
-    }
-
     private enum DdlValidationMode {
         NONE,
         WHITELIST,
diff --git a/cdc-service/src/test/java/com/xtrmetl/cdc/replication/SchemaChangeReplicaApplierLoggingTest.java b/cdc-service/src/test/java/com/xtrmetl/cdc/replication/SchemaChangeReplicaApplierLoggingTest.java
new file mode 100644
index 00000000..0c65f67b
--- /dev/null
+++ b/cdc-service/src/test/java/com/xtrmetl/cdc/replication/SchemaChangeReplicaApplierLoggingTest.java
@@ -0,0 +1,222 @@
+package com.xtrmetl.cdc.replication;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.springframework.boot.test.system.CapturedOutput;
+import org.springframework.boot.test.system.OutputCaptureExtension;
+import org.springframework.dao.DataAccessException;
+import org.springframework.jdbc.core.JdbcTemplate;
+
+import java.sql.SQLException;
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoInteractions;
+
+/**
+ * Verifies that schema-replication observability never republishes raw DDL or driver diagnostics.
+ */
+@ExtendWith(OutputCaptureExtension.class)
+class SchemaChangeReplicaApplierLoggingTest {
+
+    @Test
+    void successfulApplyLogsOnlyBoundedMetadataNotRawDdl(CapturedOutput output) {
+        JdbcTemplate jdbcTemplate = mock(JdbcTemplate.class);
+        SchemaChangeReplicaApplier applier = applier(jdbcTemplate, "none", "", "");
+        String secretLiteral = "buyer-contract-secret-8472";
+        String inputDdl = "CREATE TABLE confidential_record(secret_value text DEFAULT '" + secretLiteral + "')";
+        String executedDdl = "CREATE TABLE IF NOT EXISTS confidential_record(secret_value text DEFAULT '"
+                + secretLiteral + "')";
+
+        applier.apply(schemaTopic(), null, ddlEnvelope(inputDdl));
+
+        verify(jdbcTemplate).execute(eq(executedDdl));
+        assertSafeLogs(output, "Applied schema change DDL on replica", secretLiteral, "DEFAULT", "ddl=");
+    }
+
+    @Test
+    void blockedDdlLogsPolicyOutcomeWithoutRawStatement(CapturedOutput output) {
+        JdbcTemplate jdbcTemplate = mock(JdbcTemplate.class);
+        SchemaChangeReplicaApplier applier = applier(
+                jdbcTemplate,
+                "whitelist",
+                "CREATE TABLE,ALTER TABLE,CREATE INDEX",
+                ""
+        );
+        String secretPath = "/srv/private/buyer-contract-secret-8472";
+        String ddl = "CREATE TABLESPACE reporting LOCATION '" + secretPath + "'";
+
+        assertThrows(IllegalArgumentException.class, () -> applier.apply(schemaTopic(), null, ddlEnvelope(ddl)));
+
+        verifyNoInteractions(jdbcTemplate);
+        assertSafeLogs(output, "Blocked DDL by validation policy", secretPath, "TABLESPACE", "ddl=");
+    }
+
+    @Test
+    void multiStatementBlockLogsClassificationWithoutRawStatement(CapturedOutput output) {
+        JdbcTemplate jdbcTemplate = mock(JdbcTemplate.class);
+        SchemaChangeReplicaApplier applier = applier(jdbcTemplate, "none", "", "");
+        String secretLiteral = "buyer-contract-secret-8472";
+        String ddl = "CREATE TABLE confidential_record(id int); DROP TABLE " + secretLiteral;
+
+        assertThrows(IllegalArgumentException.class, () -> applier.apply(schemaTopic(), null, ddlEnvelope(ddl)));
+
+        verifyNoInteractions(jdbcTemplate);
+        assertSafeLogs(output, "Blocked multi-statement DDL", secretLiteral, "DROP TABLE", "ddl=");
+    }
+
+    @Test
+    void sqlCommentBlockLogsClassificationWithoutRawStatement(CapturedOutput output) {
+        JdbcTemplate jdbcTemplate = mock(JdbcTemplate.class);
+        SchemaChangeReplicaApplier applier = applier(jdbcTemplate, "none", "", "");
+        String secretLiteral = "buyer-contract-secret-8472";
+        String ddl = "CREATE TABLE confidential_record(id int) -- " + secretLiteral;
+
+        assertThrows(IllegalArgumentException.class, () -> applier.apply(schemaTopic(), null, ddlEnvelope(ddl)));
+
+        verifyNoInteractions(jdbcTemplate);
+        assertSafeLogs(
+                output,
+                "Blocked DDL containing SQL comments or NUL",
+                secretLiteral,
+                "confidential_record",
+                "ddl="
+        );
+    }
+
+    @Test
+    void nulBlockLogsClassificationWithoutRawStatement(CapturedOutput output) throws Exception {
+        JdbcTemplate jdbcTemplate = mock(JdbcTemplate.class);
+        SchemaChangeReplicaApplier applier = applier(jdbcTemplate, "none", "", "");
+        String secretLiteral = "buyer-contract-secret-8472";
+        String ddl = "CREATE TABLE confidential_record(id int) " + (char) 0 + secretLiteral;
+        String envelope = new ObjectMapper().writeValueAsString(Map.of("payload", Map.of("ddl", ddl)));
+
+        assertThrows(IllegalArgumentException.class, () -> applier.apply(schemaTopic(), null, envelope));
+
+        verifyNoInteractions(jdbcTemplate);
+        assertSafeLogs(
+                output,
+                "Blocked DDL containing SQL comments or NUL",
+                secretLiteral,
+                "confidential_record",
+                "ddl="
+        );
+    }
+
+    @Test
+    void duplicateDdlLogsOutcomeWithoutSqlOrDriverDiagnostics(CapturedOutput output) {
+        JdbcTemplate jdbcTemplate = mock(JdbcTemplate.class);
+        SchemaChangeReplicaApplier applier = applier(jdbcTemplate, "none", "", "");
+        String secretLiteral = "buyer-contract-secret-8472";
+        String inputDdl = "CREATE TABLE confidential_record(secret_value text DEFAULT '" + secretLiteral + "')";
+        String executedDdl = "CREATE TABLE IF NOT EXISTS confidential_record(secret_value text DEFAULT '"
+                + secretLiteral + "')";
+        SQLException sqlException = new SQLException(
+                "relation already exists at jdbc:postgresql://db.internal/prod?password=driver-secret",
+                "42P07"
+        );
+        DataAccessException duplicate = new DataAccessException("driver-secret", sqlException) {};
+        doThrow(duplicate).when(jdbcTemplate).execute(eq(executedDdl));
+
+        assertDoesNotThrow(() -> applier.apply(schemaTopic(), null, ddlEnvelope(inputDdl)));
+
+        verify(jdbcTemplate).execute(eq(executedDdl));
+        assertSafeLogs(
+                output,
+                "Schema change DDL already applied; skipping duplicate",
+                secretLiteral,
+                "driver-secret",
+                "jdbc:postgresql://",
+                "ddl="
+        );
+    }
+
+    @Test
+    void executionFailureLogsOutcomeWithoutSqlOrDriverDiagnostics(CapturedOutput output) {
+        JdbcTemplate jdbcTemplate = mock(JdbcTemplate.class);
+        SchemaChangeReplicaApplier applier = applier(jdbcTemplate, "none", "", "");
+        String secretLiteral = "buyer-contract-secret-8472";
+        String inputDdl = "CREATE TABLE confidential_record(secret_value text DEFAULT '" + secretLiteral + "')";
+        String executedDdl = "CREATE TABLE IF NOT EXISTS confidential_record(secret_value text DEFAULT '"
+                + secretLiteral + "')";
+        DataAccessException failure = new DataAccessException(
+                "jdbc:postgresql://db.internal/prod?password=driver-secret"
+        ) {};
+        doThrow(failure).when(jdbcTemplate).execute(eq(executedDdl));
+
+        DataAccessException thrown = assertThrows(
+                DataAccessException.class,
+                () -> applier.apply(schemaTopic(), null, ddlEnvelope(inputDdl))
+        );
+
+        assertSame(failure, thrown);
+        assertSafeLogs(
+                output,
+                "Failed to apply schema change DDL on replica",
+                secretLiteral,
+                "driver-secret",
+                "jdbc:postgresql://",
+                "ddl="
+        );
+    }
+
+    @Test
+    void malformedEventLogsOnlyStableParseClassification(CapturedOutput output) {
+        JdbcTemplate jdbcTemplate = mock(JdbcTemplate.class);
+        SchemaChangeReplicaApplier applier = applier(jdbcTemplate, "none", "", "");
+        String malformed = "{buyer-contract-secret-8472";
+
+        assertDoesNotThrow(() -> applier.apply(schemaTopic(), null, malformed));
+
+        verifyNoInteractions(jdbcTemplate);
+        assertSafeLogs(
+                output,
+                "Failed to parse Debezium schema change JSON; skipping DDL apply",
+                "buyer-contract-secret-8472",
+                "JsonParseException",
+                "ReaderBasedJsonParser"
+        );
+    }
+
+    private static SchemaChangeReplicaApplier applier(
+            JdbcTemplate jdbcTemplate,
+            String validationMode,
+            String allowedPrefixes,
+            String blockedPrefixes
+    ) {
+        return new SchemaChangeReplicaApplier(
+                jdbcTemplate,
+                new ObjectMapper(),
+                true,
+                validationMode,
+                allowedPrefixes,
+                blockedPrefixes
+        );
+    }
+
+    private static String schemaTopic() {
+        return "xtrmetl-cdc.schema-changes";
+    }
+
+    private static String ddlEnvelope(String ddl) {
+        return "{\"payload\":{\"ddl\":\"" + ddl.replace("'", "\\u0027") + "\"}}";
+    }
+
+    private static void assertSafeLogs(CapturedOutput output, String expected, String... forbidden) {
+        String logs = output.getOut() + output.getErr();
+        assertTrue(logs.contains(expected));
+        for (String value : forbidden) {
+            assertFalse(logs.contains(value), () -> "Log output exposed forbidden value: " + value);
+        }
+    }
+}