@@ -292,7 +292,9 @@ repos:
           ^java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/conformance/.*\.kt$|
           ^java-sdk/sdk/build\.gradle\.kts$|
           ^java-sdk/gradle\.properties$|
-          ^java-sdk/generated/lang-sdk/capabilities\.json$
+          ^java-sdk/generated/lang-sdk/capabilities\.json$|
+          ^scripts/ci/prek/check_java_sdk_capabilities_in_sync\.py$|
+          ^scripts/ci/prek/lang_sdk_compat_matrix\.py$
         pass_filenames: false
         require_serial: true
       - id: update-java-sdk-readme-matrix
 
@@ -17,17 +17,26 @@
  * under the License.
  */
 
-/* Dokka sizes content-table cells with `min-width: 40px` and ~31px of horizontal padding, so the
+/* Dokka sizes content-table cells with `min-width: 40px` against ~31px of horizontal padding, so the
    narrow columns of the Language SDK compatibility matrix collapse until their headers and values
-   break mid-word ("Supported", "MUST †"). Keeping every column except Dimension and Notes on a
-   single line makes each one size to its own content instead. */
+   break mid-word ("Supported", "MUST †").
+
+   These selectors necessarily apply to every content table in the API reference, because a table
+   generated from Markdown carries no class of its own. They are therefore deliberately written to
+   only *raise a floor*, never to forbid wrapping: a header stays on one line (headers are short in
+   any table), and the narrow value columns get a minimum width instead of `nowrap`. An unrelated
+   table with long prose in those columns still wraps as before rather than overflowing. */
 
 .table--container th {
   white-space: nowrap;
 }
 
+/* Tier ("SHOULD †"), Supported ("n/a") and Since ("3.3") in the compatibility matrix. */
 .table--container td:nth-child(2),
-.table--container td:nth-child(3),
+.table--container td:nth-child(3) {
+  min-width: 9ch;
+}
+
 .table--container td:nth-child(4) {
-  white-space: nowrap;
+  min-width: 6ch;
 }
@@ -22,17 +22,21 @@ import kotlinx.serialization.Serializable
 import kotlinx.serialization.encodeToString
 import kotlinx.serialization.json.Json
 
+// This model exists only to emit the build-time manifest, so it is deliberately `internal`: it is
+// not part of the SDK's published surface, and keeping it out means changing the manifest schema
+// never becomes a source/binary compatibility concern for SDK users.
+
 /** Support for a single Language SDK conformance dimension. */
 @Serializable
-data class CapabilityEntry(
+internal data class CapabilityEntry(
   val supported: Boolean,
   val since: String? = null,
   val note: String = "",
 )
 
 /** The machine-readable capability manifest serialised to java-sdk/generated/lang-sdk/capabilities.json. */
 @Serializable
-data class Document(
+internal data class CapabilityManifest(
   val sdk: String,
   @SerialName("supervisor_schema_version") val supervisorSchemaVersion: String,
   @SerialName("min_airflow_version") val minAirflowVersion: String,
@@ -57,8 +61,8 @@ private fun no(note: String = ""): CapabilityEntry = CapabilityEntry(supported =
  * The runtime terminates a task with SucceedTask, RetryTask, or TaskState (failed/removed); it does not
  * yet emit skipped, DeferTask, RescheduleTask, or AwaitInputTask.
  */
-val Capabilities: Document =
-  Document(
+internal val Capabilities: CapabilityManifest =
+  CapabilityManifest(
     sdk = "java",
     supervisorSchemaVersion = SUPERVISOR_SCHEMA_VERSION,
     minAirflowVersion = MIN_AIRFLOW_VERSION,
@@ -102,7 +106,7 @@ val Capabilities: Document =
   )
 
 /** Print the capability manifest as JSON on stdout; wired to the `:sdk:dumpCapabilities` Gradle task. */
-fun main() {
+internal fun main() {
   val json =
     Json {
       prettyPrint = true
 
@@ -41,19 +41,19 @@
 
 sys.path.insert(0, str(Path(__file__).resolve().parent))
 
-from lang_sdk_compat_matrix import AIRFLOW_ROOT_PATH, validate_capabilities
+from lang_sdk_compat_matrix import AIRFLOW_ROOT_PATH, CapabilitiesDoc, validate_capabilities
 
 JAVA_SDK = AIRFLOW_ROOT_PATH / "java-sdk"
 GRADLEW = JAVA_SDK / "gradlew"
 GRADLE_PROPERTIES = JAVA_SDK / "gradle.properties"
 CAPABILITIES_JSON = JAVA_SDK / "generated" / "lang-sdk" / "capabilities.json"
 
 
-def canonical(data: dict) -> str:
-    return json.dumps(data, indent=2) + "\n"
+def format_canonical_json(doc: CapabilitiesDoc) -> str:
+    return json.dumps(doc, indent=2) + "\n"
 
 
-def gradle_schema_version() -> str | None:
+def read_gradle_schema_version() -> str | None:
     """The ``airflowSupervisorSchemaVersion`` from gradle.properties (source of truth for the JAR)."""
     for line in GRADLE_PROPERTIES.read_text().splitlines():
         key, sep, value = line.partition("=")
@@ -62,7 +62,7 @@ def gradle_schema_version() -> str | None:
     return None
 
 
-def generate() -> dict:
+def generate_manifest() -> CapabilitiesDoc:
     """Run the Gradle dump task and parse the capability manifest it prints."""
     completed = subprocess.run(
         [str(GRADLEW), "-q", "-p", "java-sdk", ":sdk:dumpCapabilities"],
@@ -72,16 +72,18 @@ def generate() -> dict:
         check=True,
     )
     out = completed.stdout
-    # Gradle may prepend warnings; the manifest is the JSON object in the output.
-    start, end = out.find("{"), out.rfind("}")
-    if start == -1 or end == -1:
+    # Gradle may prepend warnings, so decode the first complete JSON value rather than slicing to
+    # the last "}" — a stray brace in surrounding output would make that span the wrong text.
+    start = out.find("{")
+    if start == -1:
         raise ValueError(f"no JSON object found in dumpCapabilities output:\n{out}")
-    return json.loads(out[start : end + 1])
+    doc, _ = json.JSONDecoder().raw_decode(out[start:])
+    return doc
 
 
 def main() -> int:
     if not GRADLEW.exists() or shutil.which("java") is None:
-        if os.environ.get("CI"):
+        if "CI" in os.environ:
             print("ERROR: java/gradlew is required in CI but was not found.")
             return 1
         print(
@@ -90,13 +92,13 @@ def main() -> int:
         )
         return 0
     try:
-        generated = generate()
+        generated = generate_manifest()
     except (subprocess.CalledProcessError, ValueError) as error:
         detail = getattr(error, "stderr", None) or str(error)
         print(f"ERROR: could not run `:sdk:dumpCapabilities`:\n{detail}")
         return 1
     validate_capabilities(generated, source=":sdk:dumpCapabilities", expected_sdk="java")
-    gradle_version = gradle_schema_version()
+    gradle_version = read_gradle_schema_version()
     if gradle_version is not None and generated["supervisor_schema_version"] != gradle_version:
         print(
             f"ERROR: supervisor_schema_version mismatch — the Kotlin constant emits "
@@ -110,7 +112,7 @@ def main() -> int:
         print("OK: java-sdk/generated/lang-sdk/capabilities.json is in sync with conformance.Capabilities.")
         return 0
     CAPABILITIES_JSON.parent.mkdir(parents=True, exist_ok=True)
-    CAPABILITIES_JSON.write_text(canonical(generated))
+    CAPABILITIES_JSON.write_text(format_canonical_json(generated))
     print(
         "ERROR: java-sdk/generated/lang-sdk/capabilities.json was stale and has been regenerated from "
         "conformance.Capabilities. Re-stage the file."
 
@@ -118,6 +118,8 @@ class LangSdk(TypedDict):
 # writes its manifest to generated/lang-sdk/capabilities.json under the SDK root (a generated
 # artifact kept out of the hand-authored source tree). Only the Java SDK publishes one so far; the
 # Go and TypeScript entries record where theirs go when those runtimes declare their capabilities.
+# Because of that, `capabilities_json` is a declared location and not a promise the file exists — a
+# consumer walking the whole registry must check `.exists()` before calling load_capabilities().
 LANG_SDKS: list[LangSdk] = [
     {
         "id": "go",
@@ -214,6 +216,14 @@ def _validate_entries(entries: object, *, expected: set[str], kind: str, source:
             raise CapabilitiesError(f"{source}: {kind}.{name} must be an object with a boolean 'supported'")
         if not isinstance(entry.get("since", None), (str, type(None))):
             raise CapabilitiesError(f"{source}: {kind}.{name}.since must be a string or null")
+        if not entry["supported"] and entry.get("since") is not None:
+            # "Since" means "supported since"; carrying one while unsupported is contradictory and
+            # would silently render as the not-supported placeholder. Supported *without* a version
+            # stays legal — an SDK may not know which release first shipped a dimension.
+            raise CapabilitiesError(
+                f"{source}: {kind}.{name} is not supported but carries since="
+                f"{entry['since']!r}; drop the version or mark it supported"
+            )
         if not isinstance(entry.get("note", ""), str):
             raise CapabilitiesError(f"{source}: {kind}.{name}.note must be a string")
 
 
@@ -99,6 +99,17 @@ def test_non_string_schema_version_raises(self):
         with pytest.raises(matrix.CapabilitiesError, match="supervisor_schema_version must be a string"):
             matrix.validate_capabilities(_doc(supervisor_schema_version=123), source="test")
 
+    def test_unsupported_entry_carrying_since_raises(self):
+        doc = _doc()
+        doc["capabilities"]["branching"] = _entry(False, since="3.3")
+        with pytest.raises(matrix.CapabilitiesError, match="not supported but carries since"):
+            matrix.validate_capabilities(doc, source="test")
+
+    def test_supported_entry_without_since_passes(self):
+        doc = _doc()
+        doc["states"]["success"] = _entry(True, since=None)
+        matrix.validate_capabilities(doc, source="test")
+
 
 class TestRenderMarkdownTable:
     def test_rows_cover_every_dimension(self):
@@ -127,6 +138,21 @@ def test_gated_capabilities_are_na_without_the_native_dag_gate(self):
             if cap.gated:
                 assert f"| capability: `{cap.name}` | {cap.tier} † | {matrix.NA_MARK} |" in rendered
 
+    def test_gated_capabilities_resolve_to_their_own_mark_once_the_gate_is_supported(self):
+        doc = _doc()
+        doc["capabilities"]["branching"] = _entry(False, note="no branch construct")
+        rendered = "".join(matrix.render_markdown_table(doc))
+        gated = [cap for cap in matrix.CAPABILITY_DIMENSIONS if cap.gated]
+        assert gated, "expected at least one gated capability"
+        # With the gate supported, a gated capability reports its real support, never n/a.
+        for cap in gated:
+            if cap.name != "branching":
+                assert (
+                    f"| capability: `{cap.name}` | {cap.tier} † | {matrix.SUPPORTED_MARK} | 3.3 |" in rendered
+                )
+        assert f"| capability: `branching` | SHOULD † | {matrix.UNSUPPORTED_MARK} |" in rendered
+        assert matrix.NA_MARK not in rendered.replace(matrix.LEGEND, "")
+
     def test_pipe_in_note_is_escaped(self):
         doc = _doc()
         doc["capabilities"]["xcom-read-write"] = _entry(True, since="3.3", note="read | write")