calesthio/OpenMontage · error · KlingAPIError

Unexpected Kling Classic task status

Error message

Unexpected Kling Classic task status {status!r}

What it means

Raised when poll_classic() receives a task_status value that matches neither the success, failure, nor known pending statuses (CLASSIC_PENDING_STATUSES). It is a defensive guard against Kling introducing new status enum values or returning an unexpected shape, and includes the full response payload so the unknown status can be inspected.

Solutions

1. Log the full response (attached as error.response) to see the exact status string
2. Check current Kling official API docs for the complete task_status enum
3. If the status is a new pending-like state, add it to CLASSIC_PENDING_STATUSES in tools/_kling/schemas.py
4. If it is a terminal state (cancelled/expired), add explicit handling for it instead of treating it as pending

Example fix

// schemas.py — before
CLASSIC_PENDING_STATUSES = {'submitted', 'processing'}

// schemas.py — after (after confirming against Kling docs)
CLASSIC_PENDING_STATUSES = {'submitted', 'processing', 'queuing'}

KNOWN = {'succeed', 'failed', 'submitted', 'processing'}  # mirror schemas.py

def status_is_known(status: str) -> bool:
    return status in KNOWN

from tools._kling.schemas import CLASSIC_PENDING_STATUSES, CLASSIC_SUCCESS_STATUS, CLASSIC_FAILURE_STATUS

def classify_classic_status(payload: dict) -> str:
    s = (payload.get('data') or {}).get('task_status') or (payload.get('data') or {}).get('status')
    if s == CLASSIC_SUCCESS_STATUS: return 'success'
    if s == CLASSIC_FAILURE_STATUS: return 'failure'
    if s in CLASSIC_PENDING_STATUSES: return 'pending'
    return 'unknown'

try:
    outputs = client.poll_classic(path, task_id, 'videos')
except KlingAPIError as e:
    if 'Unexpected Kling Classic task status' in str(e):
        logger.error('unknown status — full response: %s', e.response)
        # decide: extend schemas.py or treat as failure
    raise

When it happens

Trigger: Kling adds a new intermediate status (e.g. 'staging', 'queuing') not in CLASSIC_PENDING_STATUSES in schemas.py; the API returns an unusual status string for a task in an edge state (e.g. cancelled/expired tasks); payload.get('task_status') is empty and payload.get('status') carries an unexpected value.

Common situations: Kling API version drift after an upstream release; tasks that were cancelled manually in the Kling console and now report a cancel status; regional endpoints that emit slightly different status enums.

Related errors

• Unexpected Kling Turbo task status
• Kling Classic task failed
• Kling Classic task timed out after s
• Kling Turbo poll response missing data[0]
• Kling Turbo result path data[0].outputs is not a list

AI-assisted analysis of calesthio/OpenMontage@95e1c3d0ab (2026-08-15). Data as JSON: /api/errors/ff0c2594396f5862. Report an issue: GitHub.

Appendix: source

Thrown at tools/_kling/client.py:99

        timeout_seconds: int = 900,
        poll_interval: float = 5.0,
    ) -> list[dict[str, Any]]:
        deadline = time.time() + timeout_seconds
        while time.time() < deadline:
            data = self.get(f"{path.rstrip('/')}/{task_id}")
            payload = data.get("data") or {}
            status = payload.get("task_status") or payload.get("status")
            if status == CLASSIC_SUCCESS_STATUS:
                task_result = payload.get("task_result") or {}
                outputs = task_result.get(result_key) or []
                if not isinstance(outputs, list):
                    raise KlingAPIError(f"Kling Classic result path data.task_result.{result_key} is not a list")
                return outputs
            if status == CLASSIC_FAILURE_STATUS:
                message = payload.get("task_status_msg") or payload.get("message") or "Kling Classic task failed"
                raise KlingAPIError(str(message), code=payload.get("task_status"), response=data)
            if status not in CLASSIC_PENDING_STATUSES:
                raise KlingAPIError(f"Unexpected Kling Classic task status {status!r}", response=data)
            time.sleep(min(poll_interval, max(0.0, deadline - time.time())))
        raise TimeoutError(f"Kling Classic task {task_id} timed out after {timeout_seconds}s")

    def create_turbo(self, path: str, payload: dict[str, Any]) -> str:
        data = self.post(path, payload)
        task_id = ((data.get("data") or {}).get("id"))
        if not task_id:
            raise KlingAPIError(f"Kling Turbo create response missing data.id: {data}")
        return str(task_id)

    def poll_turbo(
        self,
        task_id: str,
        timeout_seconds: int = 900,
        poll_interval: float = 5.0,
    ) -> list[dict[str, Any]]:
        deadline = time.time() + timeout_seconds
        while time.time() < deadline:

View on GitHub (pinned to 95e1c3d0ab)