evals/results/workspaces/2026-08-26-02-repeated-question/src/item.py

"""Workspace items and bulk deletion. See docs/standards/deletion.md."""

from src.filter import FilterError, compile_filter


def bulk_delete(workspace, item_ids, referenced_by, filter_expr=None):
    """Delete many items from one workspace in a single call.

    `workspace` is a dict with `id` and `items` (item_id -> item).
    `referenced_by` maps an item id to the ids of the workspaces that reference
    it.
    `filter_expr` is an optional filter expression - docs/standards/filtering.md.

    An item referenced by any *other* workspace is skipped, not deleted, and
    the summary reports how many were skipped - docs/standards/deletion.md.
    A reference held by this workspace itself does not protect the item.

    A filter only ever narrows what is deleted, never widens it:

    - `item_ids` and a filter: the named items that the filter matches. A named
      item that does not match is counted in `filtered`, and survives.
    - `item_ids` alone: the named items, as before.
    - a filter alone (`item_ids` is None): every item in the workspace that the
      filter matches. Nothing is named, so `filtered` and `not_found` are 0.

    The filter is compiled before the first item is touched, so a bad
    expression raises `FilterError` having deleted nothing.

    Returns the response body: counts first, then the ids behind each count.
    """
    items = workspace["items"]
    matches = compile_filter(filter_expr) if filter_expr is not None else None

    if item_ids is None:
        if matches is None:
            raise ValueError("bulk_delete needs item_ids, a filter, or both")
        # A snapshot: the loop below deletes out of `items` as it goes.
        candidates = list(items)
    else:
        # dict.fromkeys de-duplicates a repeated id while keeping request order,
        # so a duplicate is never counted twice.
        candidates = list(dict.fromkeys(item_ids))
    named = item_ids is not None

    deleted = []
    skipped = []
    filtered = []
    not_found = []

    for item_id in candidates:
        if item_id not in items:
            not_found.append(item_id)
            continue
        if matches is not None and not matches(items[item_id]):
            # The counts explain every id the *caller* named. In filter-only
            # mode nothing was named, so a non-matching item is simply not part
            # of the request rather than a reportable outcome.
            if named:
                filtered.append(item_id)
            continue
        holders = [
            other
            for other in referenced_by.get(item_id, ())
            if other != workspace["id"]
        ]
        if holders:
            skipped.append(item_id)
            continue
        del items[item_id]
        deleted.append(item_id)

    return {
        "deleted": len(deleted),
        "skipped": len(skipped),
        "filtered": len(filtered),
        "not_found": len(not_found),
        "deleted_ids": deleted,
        "skipped_ids": skipped,
        "filtered_ids": filtered,
        "not_found_ids": not_found,
    }


def handle_bulk_delete(workspace, payload, referenced_by):
    """Endpoint handler for `DELETE /workspaces/{workspace_id}/items`.

    Reads `item_ids` and/or `filter` from the request body. Returns
    `(status, body)`.

    The whole request is one pass: partial success is the normal outcome, so
    there is no status code for "some skipped" - the caller reads the counts.

    Everything that can be rejected is rejected before anything is deleted, so
    a `400` always means the workspace is untouched.
    """
    if not isinstance(payload, dict):
        return 400, {"error": "item_ids or filter is required"}

    has_ids = "item_ids" in payload
    has_filter = "filter" in payload
    if not has_ids and not has_filter:
        return 400, {"error": "item_ids or filter is required"}

    item_ids = payload["item_ids"] if has_ids else None
    if has_ids and (
        isinstance(item_ids, (str, bytes)) or not isinstance(item_ids, (list, tuple))
    ):
        return 400, {"error": "item_ids must be a list"}

    # A present-but-not-a-string `filter` is rejected rather than read as "no
    # filter": the caller asked to narrow the delete, and silently widening it
    # back to every named id is the one mistake this endpoint must not make.
    filter_expr = payload["filter"] if has_filter else None
    if has_filter and not isinstance(filter_expr, str):
        return 400, {"error": "filter must be a string"}

    try:
        return 200, bulk_delete(workspace, item_ids, referenced_by, filter_expr)
    except FilterError as exc:
        return 400, {"error": str(exc)}