prefer-coordinated-state-owner · Python

Why

Repeated same-key writes and removals in separate dictionaries distribute a possible entry-lifetime invariant across writers. One typed owner can hold the coordinated values. This is structural evidence for review, not proof of a defect or a shared domain lifetime.

Fix

Review the reported fields, writers and readers. If entries have one lifetime, use one dictionary of typed owner records retaining each value's annotation. Preserve task identities, start/stop order, exception observation, cancellation and joining. Keep independent outcome reporting, historical observations, partial results and independently removed entries separate.

Examples

Before — flagged Observer and playback entries repeat paired bookkeeping

import asyncio


class Playback:
    def __init__(self) -> None:
        self._observers: dict[str, asyncio.Task[None]] = {}
        self._playbacks: dict[str, asyncio.Task[bool]] = {}

    def start(
        self, key: str, observer: asyncio.Task[None], playback: asyncio.Task[bool]
    ) -> None:
        self._observers[key] = observer
        self._playbacks[key] = playback

    def finish(self, key: str) -> None:
        self._observers.pop(key, None)
        self._playbacks.pop(key, None)

After — preferred One typed entry retains both task identities

import asyncio
from dataclasses import dataclass


@dataclass(frozen=True)
class ActivePlayback:
    observer: asyncio.Task[None]
    playback: asyncio.Task[bool]


class Playback:
    def __init__(self) -> None:
        self._active: dict[str, ActivePlayback] = {}

    def start(
        self, key: str, observer: asyncio.Task[None], playback: asyncio.Task[bool]
    ) -> None:
        self._active[key] = ActivePlayback(observer, playback)

    def finish(self, key: str) -> None:
        self._active.pop(key, None)