JuliusBrussee/caveman · error · MiddlewareError

unknown_capability

unknown_capability

Error message

unknown_capability

What it means

MiddlewareError raised by capabilities() when a capability manifest passes schema-version checks but contains a transform entry that is structurally invalid: duplicate transform_id, bad token fields, non-list eligible_segment_kinds, recovery not in ("exact_ccr","none"), or deterministic is not literally True. Only vetted, deterministic, recoverable transforms are accepted.

Solutions

1. Fix the offending transform entry: make transform_id unique, recovery exactly "exact_ccr" or "none", deterministic literally true, eligible_segment_kinds a list
2. Remove duplicate transform entries or merge them under one id
3. If the transform is legitimately new, update the validator/contract rather than loosening the manifest
4. Pre-validate the manifest with the check below before calling

Example fix

# before
{"transform_id": "t1", "recovery": "best_effort", "deterministic": false, ...}
# after
{"transform_id": "t1", "recovery": "none", "deterministic": True, "eligible_segment_kinds": ["prose"]}

def transform_entry_ok(t) -> bool:
    return (isinstance(t.get("transform_id"), str) and t["transform_id"]
            and t.get("recovery") in ("exact_ccr", "none")
            and t.get("deterministic") is True
            and isinstance(t.get("eligible_segment_kinds"), list))

def manifest_transforms_ok(doc) -> bool:
    ids = [t.get("transform_id") for t in doc.get("transforms", [])]
    return len(ids) == len(set(ids)) and all(transform_entry_ok(t) for t in doc["transforms"])

def is_deterministic_transform(t) -> bool:
    return isinstance(t, dict) and t.get("deterministic") is True and t.get("recovery") in ("exact_ccr", "none")

try:
    caps = client.capabilities(manifest)
except MiddlewareError as e:
    if e.args[0] == "unknown_capability":
        log.error("a transform entry violates the capability contract")
    raise

When it happens

Trigger: Calling capabilities() (or any flow that validates the manifest) with a transforms array containing: a repeated transform_id, recovery set to e.g. "best_effort", deterministic omitted/False, or eligible_segment_kinds as a string instead of a list.

Common situations: Hand-editing a capability manifest and typo-ing recovery; copying a transform entry without changing transform_id; a provider shipping a new transform type the validator doesn't whitelist; schema drift between middleware versions.

Understand the failure class

Background: Schema validation failed / invalid input schema: payload rejected because its shape doesn't match the expected schema — this error's family across 28 libraries.

Related errors

• unsupported_version
• artifact_id is required
• ASGI context resolver or request bounds are invalid
• assemble requires model and session_id
• assembly slot id must be non-empty and unique

AI-assisted analysis of JuliusBrussee/caveman@3ee70a1026 (2026-09-20). Data as JSON: /api/errors/478ccceab04b1cc5. Report an issue: GitHub.

Appendix: source

Thrown at packages/sdk/python/caveman_cloud/middleware/validate.py:47

    if not all(token(v) for v in values):
        raise MiddlewareError("invalid_scope")
    return json.dumps(values, separators=(",", ":"))


def capabilities(value: Any) -> dict[str, Any]:
    try:
        if (type(value["schema_version"]) is not int or value["schema_version"] != 1 or not token(value["policy_revision"])
                or not isinstance(value["runtime_build"], str) or value["mode"] not in ("record", "compress")
                or not isinstance(value["transforms"], list) or len(value["transforms"]) > 128
                or not all(integer(value["limits"][k]) and value["limits"][k] > 0 for k in ("deadline_ms", "request_bytes", "segment_bytes", "page_bytes"))
                or not integer(value["retention_seconds"]) or type(value["recovery"]) is not bool
                or type(value["persistent"]) is not bool):
            raise ValueError()
        seen = set()
        for t in value["transforms"]:
            if (not token(t["transform_id"]) or t["transform_id"] in seen or not token(t["implementation_version"])
                    or not isinstance(t["eligible_segment_kinds"], list) or t["recovery"] not in ("exact_ccr", "none") or t["deterministic"] is not True):
                raise MiddlewareError("unknown_capability")
            seen.add(t["transform_id"])
        return value
    except (KeyError, TypeError, ValueError) as error:
        raise MiddlewareError("unsupported_version") from None


def plan(value: Any, request: dict, input_digest: str, caps: dict) -> dict:
    try:
        p = value
        if (type(p["schema_version"]) is not int or p["schema_version"] != 1 or p["request_id"] != request["request_id"] or p["input_digest"] != input_digest
                or p["policy_revision"] != request["policy"]["revision"] or not digest(p["replacement_set_id"])
                or p["status"] not in ("optimized", "bypassed", "record") or not token(p["reason"])
                or not isinstance(p["replacements"], list) or not isinstance(p["skipped"], list)):
            raise ValueError()
        m = p["measurement"]
        recovery = p["recovery"]
        if (m["basis"] != "inferred" or m["scope"] != "segment" or m["verified_saved_usd"] != 0
                or not isinstance(m["tokenizer"], str)

View on GitHub (pinned to 3ee70a1026)