"""A deterministic lost-response example. Run with Python 3.10 or newer.

No network, LLM, real tickets, or persistent storage is involved. The sink is
single-threaded and survives the simulated client failure. Its in-memory
deduplication record and ticket write are treated as one indivisible operation.
This illustrates operation identity; it is not a production implementation or
a benchmark of an agent, database, or workflow framework.
"""

from dataclasses import dataclass, field


@dataclass
class TicketSink:
    deduplicate: bool
    tickets: list[str] = field(default_factory=list)
    completed: dict[str, tuple[str, str]] = field(default_factory=dict)

    def create(self, operation_id: str, title: str, lose_response: bool = False) -> str:
        if self.deduplicate and operation_id in self.completed:
            original_title, ticket_id = self.completed[operation_id]
            if title != original_title:
                raise ValueError("Same operation key with different parameters")
            return ticket_id

        ticket_id = f"ticket-{len(self.tickets) + 1}"
        self.tickets.append(title)
        if self.deduplicate:
            self.completed[operation_id] = (title, ticket_id)

        # The remote effect exists before the client observes a failure.
        if lose_response:
            raise TimeoutError("The reply was lost after the ticket was created")
        return ticket_id


def run_case(label: str, deduplicate: bool, retry_key: str) -> int:
    sink = TicketSink(deduplicate=deduplicate)
    operation_id = "approved-operation-1"
    title = "Fix the broken export"
    try:
        sink.create(operation_id, title, lose_response=True)
    except TimeoutError:
        # This handler knows only that no answer arrived. It cannot infer that
        # the effect failed. We deliberately retry to expose the difference.
        sink.create(retry_key, title)
    count = len(sink.tickets)
    print(f"{label}: {count} ticket{'s' if count != 1 else ''}")
    return count


if __name__ == "__main__":
    assert run_case("No provider deduplication", False, "approved-operation-1") == 2
    assert run_case("Same operation key", True, "approved-operation-1") == 1
    assert run_case("New key after replanning", True, "new-tool-call-2") == 2