# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/which-invariants-prevent-artifact-stream-and-push-callback-divergence-in-a2a-v1-0-task-lifecycles.md
# Remnant Warrant:    https://corpusfetch.com/c/which-invariants-prevent-artifact-stream-and-push-callback-divergence-in-a2a-v1-0-task-lifecycles.warrant.json
# Markdown SHA-256:   0a532053d601817c6c0a1364d64ead81665e9e2bac8978330543293c6f011861
# Code AST SHA-256:   f81a21ef25e312af156e1f3ddce88885e667d14ffa883b1b18c4acab5541a4e1
# Verified Classes:   ArtifactFlushInvariantError, TaskChannelState
# Defensive Guards:   2 fail-closed invariants (zero external deps)

class ArtifactFlushInvariantError(RuntimeError):
    """Raised when a terminal task state is about to be published on the
    push-notification channel before the final artifact chunk has been
    enqueued on every active delivery channel (A2A v1.0 invariant)."""
    pass


class TaskChannelState:
    def __init__(self, task_id: str):
        self.task_id = task_id
        self.final_chunk_enqueued = {"stream": False, "push": False}
        self.state = "working"

    def mark_chunk_enqueued(self, channel: str, last_chunk: bool) -> None:
        if channel not in self.final_chunk_enqueued:
            raise ArtifactFlushInvariantError(
                f"unknown delivery channel {channel!r} for task {self.task_id}"
            )
        if last_chunk:
            self.final_chunk_enqueued[channel] = True

    def assert_terminal_publishable(self, target_state: str) -> None:
        if target_state not in ("completed", "failed", "canceled"):
            return  # non-terminal transitions are not gated
        missing = [ch for ch, done in self.final_chunk_enqueued.items() if not done]
        if missing:
            raise ArtifactFlushInvariantError(
                f"task {self.task_id}: refusing to publish terminal state "
                f"{target_state!r}; final artifact chunk not enqueued on {missing}"
            )
        self.state = target_state


def publish_terminal_callback(task: TaskChannelState, target_state: str) -> None:
    # Enforces enqueue-before-terminal ordering across stream and push channels.
    task.assert_terminal_publishable(target_state)
    # ... dispatch push callback carrying target_state ...