fix(cdc): expose truthful target execution capabilities

cdc-service/src/main/java/com/xtrmetl/cdc/spi/CdcTargetConnector.java

@@ -8,6 +8,32 @@
  */
 public interface CdcTargetConnector extends AutoCloseable {
 
+    /**
+     * Describes how the currently shipped product path delivers events to a target.
+     */
+    enum DeliveryMode {
+        /** Raw Debezium envelopes are published by {@code CdcService} to Kafka. */
+        RAW_DEBEZIUM_KAFKA,
+        /** Processed-data rows are applied by the dedicated JDBC replica pipeline. */
+        PROCESSED_DATA_JDBC_REPLICA,
+        /** Canonical records are delivered directly through this SPI's {@link #write(List)} method. */
+        CANONICAL_RECORD_SPI
+    }
+
+    /**
+     * Truthful execution metadata for a target connector.
+     *
+     * @param productPathLive whether mightyETL currently has a live product path for the target
+     * @param canonicalWriteSupported whether {@link #write(List)} is wired for canonical records
+     * @param deliveryMode the execution boundary used by the live product path
+     */
+    record Capabilities(
+            boolean productPathLive,
+            boolean canonicalWriteSupported,
+            DeliveryMode deliveryMode
+    ) {
+    }
+
     String id();
 
     String displayName();
@@ -17,6 +43,13 @@ public interface CdcTargetConnector extends AutoCloseable {
      */
     boolean scaffoldOnly();
 
+    /**
+     * Returns execution metadata without conflating a live legacy/product path with SPI write support.
+     *
+     * @return immutable target capability metadata
+     */
+    Capabilities capabilities();
+
     void validate(Map<String, String> config);
 
     /**

cdc-service/src/main/java/com/xtrmetl/cdc/spi/JdbcReplicaCdcTargetConnector.java

@@ -25,6 +25,16 @@ public boolean scaffoldOnly() {
         return false;
     }
 
+    /**
+     * Reports that replica apply is live through the processed-data applier while canonical SPI writes remain unwired.
+     *
+     * @return immutable JDBC replica target execution metadata
+     */
+    @Override
+    public Capabilities capabilities() {
+        return new Capabilities(true, false, DeliveryMode.PROCESSED_DATA_JDBC_REPLICA);
+    }
+
     @Override
     public void validate(Map<String, String> config) {
         if (config == null) {

cdc-service/src/main/java/com/xtrmetl/cdc/spi/KafkaCdcTargetConnector.java

@@ -26,6 +26,16 @@ public boolean scaffoldOnly() {
         return false;
     }
 
+    /**
+     * Reports that Kafka is live through the raw Debezium path while canonical SPI writes remain unwired.
+     *
+     * @return immutable Kafka target execution metadata
+     */
+    @Override
+    public Capabilities capabilities() {
+        return new Capabilities(true, false, DeliveryMode.RAW_DEBEZIUM_KAFKA);
+    }
+
     @Override
     public void validate(Map<String, String> config) {
         if (config == null) {

cdc-service/src/test/java/com/xtrmetl/cdc/spi/CdcTargetCapabilityTest.java

@@ -0,0 +1,43 @@
+package com.xtrmetl.cdc.spi;
+
+import org.junit.jupiter.api.Test;
+
+import java.lang.reflect.Method;
+import java.util.Arrays;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+class CdcTargetCapabilityTest {
+
+    @Test
+    void liveTargetsDiscloseTheirActualExecutionBoundary() throws Exception {
+        Method capabilitiesMethod = Arrays.stream(CdcTargetConnector.class.getMethods())
+                .filter(method -> method.getName().equals("capabilities"))
+                .findFirst()
+                .orElse(null);
+
+        assertTrue(capabilitiesMethod != null,
+                "CDC target SPI must expose capabilities instead of overloading scaffoldOnly");
+
+        assertCapabilities(
+                capabilitiesMethod.invoke(new KafkaCdcTargetConnector()),
+                "RAW_DEBEZIUM_KAFKA"
+        );
+        assertCapabilities(
+                capabilitiesMethod.invoke(new JdbcReplicaCdcTargetConnector()),
+                "PROCESSED_DATA_JDBC_REPLICA"
+        );
+    }
+
+    private static void assertCapabilities(Object capabilities, String expectedDeliveryMode) throws Exception {
+        Method productPathLive = capabilities.getClass().getMethod("productPathLive");
+        Method canonicalWriteSupported = capabilities.getClass().getMethod("canonicalWriteSupported");
+        Method deliveryMode = capabilities.getClass().getMethod("deliveryMode");
+
+        assertTrue((Boolean) productPathLive.invoke(capabilities));
+        assertFalse((Boolean) canonicalWriteSupported.invoke(capabilities));
+        assertEquals(expectedDeliveryMode, ((Enum<?>) deliveryMode.invoke(capabilities)).name());
+    }
+}