# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/when-must-an-a2a-v1-0-client-treat-a-task-as-terminal-despite-a-still-open-sse-artifact-stream.md
# Remnant Warrant:    https://corpusfetch.com/c/when-must-an-a2a-v1-0-client-treat-a-task-as-terminal-despite-a-still-open-sse-artifact-stream.warrant.json
# Markdown SHA-256:   af111f9209e58e62fbe6b29a036df147599c31ec92c0d727d674ea6975607246
# Code AST SHA-256:   443a786a1f26f35e862e0c8b1b11930fb90d415ebd24cce014c3ca4a670e8727
# Verified Classes:   ArtifactAbandonedError, ArtifactState, ArtifactTracker
# Defensive Guards:   2 fail-closed invariants (zero external deps)

from __future__ import annotations

from dataclasses import dataclass, field
from enum import Enum
from typing import Iterable


class ArtifactAbandonedError(RuntimeError):
    """Raised when an artifact is consumed without a lastChunk:true frame
    before the task reached a terminal state."""


class ArtifactState(Enum):
    OPEN = "open"
    COMPLETE = "complete"
    ABANDONED = "abandoned"


TERMINAL_STATES = {"completed", "failed", "canceled", "rejected"}


@dataclass
class ArtifactTracker:
    task_id: str
    _state: dict[str, ArtifactState] = field(default_factory=dict)
    _task_state: str = "submitted"

    def on_artifact_update(self, artifact_id: str, last_chunk: bool) -> None:
        if self._task_state in TERMINAL_STATES:
            # Late artifact frame after terminal status: protocol violation
            # or resubscribe replay. Do not resurrect a terminal task.
            raise ArtifactAbandonedError(
                f"artifact {artifact_id} arrived after task {self.task_id} "
                f"reached terminal state {self._task_state}"
            )
        if last_chunk:
            self._state[artifact_id] = ArtifactState.COMPLETE
        else:
            self._state.setdefault(artifact_id, ArtifactState.OPEN)

    def on_status_update(self, state: str, final: bool) -> None:
        self._task_state = state
        if state in TERMINAL_STATES:
            for aid, st in self._state.items():
                if st is ArtifactState.OPEN:
                    self._state[aid] = ArtifactState.ABANDONED

    def require_complete(self, artifact_id: str) -> None:
        st = self._state.get(artifact_id, ArtifactState.ABANDONED)
        if st is not ArtifactState.COMPLETE:
            raise ArtifactAbandonedError(
                f"artifact {artifact_id} on task {self.task_id} is {st.value}; "
                f"task state={self._task_state}. Refusing to emit partial data."
            )

    def complete_artifacts(self) -> Iterable[str]:
        return [a for a, s in self._state.items() if s is ArtifactState.COMPLETE]