Function read_pointer

src/bridge/bridge_pointer.py:109–192

Return the pointer for ``working_dir`` if one exists and is valid for the current host, else ``None``. Returns ``None`` for any of: file missing, JSON malformed, ``schema_version`` mismatch, ``machine_name`` mismatch, ``dir`` mismatch (the file was written by another working tree us

(
    working_dir: str, *, machine_name: str,
)

def read_pointer(
    working_dir: str, *, machine_name: str,
) -> BridgePointer | None:
    """Return the pointer for ``working_dir`` if one exists and is
    valid for the current host, else ``None``.

    Returns ``None`` for any of: file missing, JSON malformed,
    ``schema_version`` mismatch, ``machine_name`` mismatch, ``dir``
    mismatch (the file was written by another working tree using a
    shared HOME), or any required field missing/wrong type.
    """
    path = _pointer_path(working_dir)
    try:
        with open(path, 'r', encoding='utf-8') as fh:
            raw = fh.read()
    except FileNotFoundError:
        return None
    except OSError as err:
        logger.warning(
            '[bridge:pointer] read %s failed: %s — treating as absent',
            path, err,
        )
        return None
    try:
        data = json.loads(raw)
    except json.JSONDecodeError as err:
        logger.warning(
            '[bridge:pointer] %s is not valid JSON (%s) — treating as absent',
            path, err,
        )
        return None
    if not isinstance(data, dict):
        return None
    if data.get('schema_version') != _SCHEMA_VERSION:
        logger.info(
            '[bridge:pointer] schema_version mismatch in %s '
            '(file=%r, expected=%r) — treating as absent',
            path, data.get('schema_version'), _SCHEMA_VERSION,
        )
        return None
    # Required string fields.
    try:
        bridge_id = str(data['bridge_id'])
        environment_id = str(data['environment_id'])
        file_machine = str(data['machine_name'])
        file_dir = str(data['dir'])
        created_at_ms = int(data['created_at_ms'])
        updated_at_ms = int(data['updated_at_ms'])
    except (KeyError, TypeError, ValueError) as err:
        logger.info(
            '[bridge:pointer] %s missing or malformed required fields '
            '(%s) — treating as absent', path, err,
        )
        return None
    session_id_raw = data.get('session_id')
    session_id = (
        str(session_id_raw) if isinstance(session_id_raw, str) else None
    )