Function write_pointer

src/bridge/bridge_pointer.py:195–246

Write/overwrite the pointer for ``working_dir``. ``created_at_ms`` defaults to the current time on first write; callers preserving a pre-existing pointer's identity should pass its ``created_at_ms`` so the field doesn't reset on every update. Writes through a tmpfile + ``os.replace

(
    working_dir: str, *,
    bridge_id: str,
    environment_id: str,
    session_id: str | None,
    machine_name: str,
    created_at_ms: int | None = None,
)

def write_pointer(
    working_dir: str, *,
    bridge_id: str,
    environment_id: str,
    session_id: str | None,
    machine_name: str,
    created_at_ms: int | None = None,
) -> None:
    """Write/overwrite the pointer for ``working_dir``.

    ``created_at_ms`` defaults to the current time on first write;
    callers preserving a pre-existing pointer's identity should pass
    its ``created_at_ms`` so the field doesn't reset on every update.

    Writes through a tmpfile + ``os.replace`` for atomicity — a crash
    during the write cannot leave a half-written pointer that
    ``read_pointer`` then parses incorrectly.
    """
    path = _pointer_path(working_dir)
    now = _now_ms()
    pointer = BridgePointer(
        bridge_id=bridge_id,
        environment_id=environment_id,
        session_id=session_id,
        machine_name=machine_name,
        dir=working_dir,
        created_at_ms=created_at_ms if created_at_ms is not None else now,
        updated_at_ms=now,
    )
    try:
        os.makedirs(os.path.dirname(path), exist_ok=True)
    except OSError as err:
        logger.warning(
            '[bridge:pointer] mkdir(%s) failed: %s — pointer not written',
            os.path.dirname(path), err,
        )
        return
    tmp_path = f'{path}.tmp'
    try:
        with open(tmp_path, 'w', encoding='utf-8') as fh:
            json.dump(pointer.to_json(), fh, indent=2)
        os.replace(tmp_path, path)
    except OSError as err:
        logger.warning(
            '[bridge:pointer] write %s failed: %s — pointer not written',
            path, err,
        )
        # Best-effort cleanup of the tmpfile.
        try:
            os.unlink(tmp_path)
        except OSError:
            pass


def clear_pointer(working_dir: str) -> None: