fix: recognize exact status-capacity messages in string responses

docs/knowledge-model-policy.md

@@ -376,8 +376,10 @@ deadline for reading the response.
 Status POSTs are never retried by the transport. An HTTP 422 status rejection is
 terminal only for the current runner invocation: `status-rejected` remains an
 execution error, not declaration noncompliance or a successful invalidation.
-An explicit recognized status-capacity response is `budget-saturated`; a count of
-900 or an unknown 422 is not capacity proof. Other transient write failures retain
+An explicit recognized status-capacity response is `budget-saturated`: the fixed
+full capacity message is recognized either directly in the `errors` string or in
+a structured `Status`/`custom` array entry. An arbitrary string, substring match,
+count of 900, or unknown 422 is not capacity proof. Other transient write failures retain
 the bounded outer retries. A fresh invocation may retry after remediation.
 The previous remote status may remain green or pending when a write is rejected;
 no failure status is claimed to have been published. Diagnostics and retry control

tests/test_knowledge_policy_http.py

@@ -47,6 +47,36 @@ def test_explicit_status_capacity_reason_is_recognized(self):
         self.assertIn('"reason": "status-capacity"', str(error))
         self.assertNotIn("This SHA", str(error))
 
+    def test_exact_string_capacity_is_recognized_but_remains_execution_failure(self):
+        body = {"message": "Validation Failed", "errors":
+                "This SHA and context has reached the maximum number of statuses."}
+        error = policy.GitHubHTTPError(HTTPError("https://example.invalid", 422, "", {},
+                                              io.BytesIO(json.dumps(body).encode())))
+        self.assertEqual(error.reason, "status-capacity")
+        self.assertNotIn(body["errors"], str(error))
+        result = policy.status_write_failure(error, {"description": "fixture"})
+        result.update(pull_request=1, rerun=result["kind"] in policy.RERUN_KINDS)
+        self.assertEqual(result["kind"], "budget-saturated")
+        self.assertEqual(result["state"], "failure")
+        self.assertFalse(result["rerun"])
+        self.assertEqual(policy.exit_code([result]), policy.EXIT_NONCOMPLIANT)
+        self.assertEqual(policy.summarize([result])["execution"], "failed")
+        self.assertEqual(policy.summarize([result])["noncompliant_pull_requests"], [])
+
+    def test_capacity_string_near_misses_remain_unknown_and_redacted(self):
+        capacity = "This SHA and context has reached the maximum number of statuses."
+        for errors in ("FAKE_SECRET", capacity + " FAKE_SECRET", "FAKE_SECRET " + capacity,
+                       capacity.replace("maximum", "minimum"), capacity[:-1]):
+            with self.subTest(errors=errors):
+                body = {"message": capacity, "errors": errors}
+                error = policy.GitHubHTTPError(HTTPError("https://example.invalid", 422, "", {},
+                                                      io.BytesIO(json.dumps(body).encode())))
+                self.assertEqual(error.reason, "unknown")
+                self.assertNotIn("FAKE_SECRET", str(error))
+                self.assertNotIn("This SHA", str(error))
+                self.assertEqual(policy.status_write_failure(
+                    error, {"description": "fixture"})["kind"], "status-rejected")
+
     def test_unknown_validation_is_redacted_and_not_inferred_as_capacity(self):
         capacity = "This SHA and context has reached the maximum number of statuses."
         for item in ({"resource": "Status", "code": "custom", "message": "FAKE_SECRET"},

tools/check_knowledge_model_policy.py

@@ -350,12 +350,16 @@ def __init__(self, error):
             # HTTP 422, an arbitrary validation message, or the locally counted history.
             capacity_message = "this sha and context has reached the maximum number of statuses."
             errors = body.get("errors", []) if isinstance(body, dict) else []
-            capacity = isinstance(errors, list) and any(
-                isinstance(item, dict) and item.get("resource") == "Status"
-                and item.get("code") == "custom"
-                and isinstance(item.get("message"), str)
-                and item["message"].strip().lower() == capacity_message
-                for item in errors)
+            # GitHub may return the same capacity diagnostic directly as `errors`
+            # rather than in a Status/custom array. Match only this fixed full text.
+            capacity = (isinstance(errors, str) and
+                        errors.strip().lower() == capacity_message) or (
+                isinstance(errors, list) and any(
+                    isinstance(item, dict) and item.get("resource") == "Status"
+                    and item.get("code") == "custom"
+                    and isinstance(item.get("message"), str)
+                    and item["message"].strip().lower() == capacity_message
+                    for item in errors))
             self.reason = "status-capacity" if capacity else "unknown"
             detail["reason"] = self.reason
             # Structural telemetry only: fixed types/booleans and counts bounded to