000-prompt.md

This is the exact prompt I sent to R1, accidental typos and all.

---

I need your help cleaning up "abandoned" threadgates (restricted reply requests) on my account. Write a Python program to do it. `requests` and `rich` are available. It should be fast and performant and kind on both my resources and Bluesky's.

Here's a HTTPie example of how to acquire an access JWT and the DID:

http \
  POST \
  https://bsky.social/xrpc/com.atproto.server.createSession \
  identifier=USERNAME.bsky.social \
  password=APP_SPECIFIC_PASSWORD

(You should prompt the user for the username and password, obviously.)

You will find the access JWT in .accessJWT and the DID in .did.

Then, get the threadgates:

http \
     https://bsky.social/xrpc/com.atproto.repo.listRecords \
     repo==$DID \
     collection==app.bsky.feed.threadgate \
     limit==100

The response will look like this:

{
    "cursor": "3l7xw73h5if2y",
    "records": [
        {
            "uri": "at://whatever/app.bsky.feed.threadgate/my-threadgate"
            "value": {
                "post": "at://whatever/app.bsky.feed.post/3lgti6bpe3s2c"
            }
        }
    ]
}

(You should keep requesting pages using the cursor query parameter.)

Then you need to check if the threadgate is "abandoned" by checking if the post still exists:

http \
    -A bearer \
    -a $ACCESS_JWT \
     https://bsky.social/xrpc/com.atproto.repo.getRecord \
     repo==$DID \
     collection==app.bsky.feed.post \
     rkey=3lgti6bpe3s2c

This will return 200 OK if the post is still up or 400 Bad Request if it's been deleted.

If the post has been deleted, delete the threadgate:

http \
    -A bearer \
    -a $ACCESS_JWT \
     https://bsky.social/xrpc/com.atproto.repo.deleteRecord \
     repo==$DID \
     collection==app.bsky.feed.threadgate \
     rkey=my-threadgate

Repeat this process for all abandoned threadgates.

Make sure to make the script (both the code and its output) delightful, clean and usable, and provide insightful and beautiful logs at every step.

001-delete-abandoned-threadgates-r1.py

# /// script
# requires-python = ">=3.13"
# dependencies = [
#   "requests",
#   "rich"
# ]
# ///


import requests
from rich.console import Console
from rich.progress import Progress
from rich.table import Table
from rich import box
import getpass

# Configure rich console
console = Console()
BASE_URL = "https://bsky.social/xrpc/"

def create_session(username, password):
    """Create a Bluesky session and return access token + DID"""
    url = BASE_URL + "com.atproto.server.createSession"
    response = requests.post(url, json={"identifier": username, "password": password})
    response.raise_for_status()
    data = response.json()
    return data["accessJwt"], data["did"]

def list_threadgates(did, access_jwt):
    """List all threadgates with pagination"""
    headers = {"Authorization": f"Bearer {access_jwt}"}
    records = []
    cursor = None

    with console.status("[bold cyan]Fetching threadgates...[/]", spinner="dots"):
        while True:
            params = {"repo": did, "collection": "app.bsky.feed.threadgate", "limit": 100}
            if cursor:
                params["cursor"] = cursor

            response = requests.get(
                BASE_URL + "com.atproto.repo.listRecords",
                params=params,
                headers=headers
            )
            response.raise_for_status()
            data = response.json()
            records.extend(data.get("records", []))
            cursor = data.get("cursor")
            if not cursor:
                break

    return records

def post_exists(did, post_rkey, access_jwt):
    """Check if a post exists"""
    headers = {"Authorization": f"Bearer {access_jwt}"}
    params = {"repo": did, "collection": "app.bsky.feed.post", "rkey": post_rkey}

    response = requests.get(
        BASE_URL + "com.atproto.repo.getRecord",
        params=params,
        headers=headers
    )
    return response.status_code == 200

def delete_threadgate(did, rkey, access_jwt):
    """Delete a threadgate record"""
    headers = {"Authorization": f"Bearer {access_jwt}"}
    data = {"repo": did, "collection": "app.bsky.feed.threadgate", "rkey": rkey}

    response = requests.post(
        BASE_URL + "com.atproto.repo.deleteRecord",
        json=data,
        headers=headers
    )
    response.raise_for_status()

def main():
    console.print("\n[bold magenta]✨ Bluesky Threadgate Cleanup Tool ✨[/bold magenta]\n")

    # Get credentials
    username = input("Enter your username (e.g., user.bsky.social): ")
    password = getpass.getpass("Enter your app-specific password: ")

    try:
        # Create session
        with console.status("[bold green]Authenticating...[/]"):
            access_jwt, did = create_session(username, password)
        console.print(f"✅ [bold green]Authenticated as [cyan]{did}[/cyan][/bold green]\n")

        # Get all threadgates
        threadgates = list_threadgates(did, access_jwt)
        if not threadgates:
            console.print("[bold yellow]🎉 No threadgates found! Nothing to clean up.[/]")
            return

        console.print(f"🔍 Found [bold cyan]{len(threadgates)}[/] threadgates to check\n")

        # Process threadgates
        deleted_count = 0
        with Progress() as progress:
            task = progress.add_task("[cyan]Processing threadgates...", total=len(threadgates))

            for record in threadgates:
                try:
                    # Extract keys
                    tg_uri = record["uri"]
                    tg_rkey = tg_uri.split("/")[-1]
                    post_uri = record["value"]["post"]
                    post_rkey = post_uri.split("/")[-1]

                    # Check post existence
                    if not post_exists(did, post_rkey, access_jwt):
                        console.print(f"  🗑️ [yellow]Deleting abandoned threadgate [bold]{tg_rkey}[/][/]")
                        delete_threadgate(did, tg_rkey, access_jwt)
                        deleted_count += 1
                        progress.console.print(f"  ✅ [red]Deleted [bold]{tg_rkey}[/][/]")
                    else:
                        progress.console.print(f"  🔗 [grey58]Keeping active threadgate [bold]{tg_rkey}[/][/]")

                except Exception as e:
                    console.print(f"[red]⚠️ Error processing {tg_rkey}: {str(e)}[/]")
                finally:
                    progress.update(task, advance=1)

        # Show summary
        console.print("\n[bold magenta]🏁 Cleanup Complete![/]")
        table = Table(box=box.ROUNDED, show_header=False, padding=(0, 2))
        table.add_row("Total Threadgates Checked", f"[bold]{len(threadgates)}[/]")
        table.add_row("Abandoned Threadgates Deleted", f"[bold red]{deleted_count}[/]")
        table.add_row("Active Threadgates Remaining", f"[bold green]{len(threadgates) - deleted_count}[/]")
        console.print(table)

    except Exception as e:
        console.print(f"[bold red]🚨 Critical Error: {str(e)}[/]")
        raise

if __name__ == "__main__":
    main()

002-claude.py

# /// script
# requires-python = ">=3.13"
# dependencies = [
#     "click",
#     "requests",
#     "rich",
# ]
# ///

import asyncio
from dataclasses import dataclass
from typing import Optional
import click
import requests
from rich.console import Console
from rich.progress import Progress, SpinnerColumn, TextColumn
from rich.prompt import Prompt

console = Console()

@dataclass
class BlueskySession:
    access_jwt: str
    did: str

class BlueskyClient:
    def __init__(self, session: BlueskySession):
        self.session = session
        self.base_url = "https://bsky.social/xrpc"
        
    def list_threadgates(self, cursor: Optional[str] = None) -> dict:
        """Get a page of threadgates."""
        params = {
            "repo": self.session.did,
            "collection": "app.bsky.feed.threadgate",
            "limit": 100,
        }
        if cursor:
            params["cursor"] = cursor
            
        response = requests.get(
            f"{self.base_url}/com.atproto.repo.listRecords",
            params=params
        )
        response.raise_for_status()
        return response.json()
    
    def check_post_exists(self, rkey: str) -> bool:
        """Check if a post still exists."""
        response = requests.get(
            f"{self.base_url}/com.atproto.repo.getRecord",
            params={
                "repo": self.session.did,
                "collection": "app.bsky.feed.post",
                "rkey": rkey
            },
            headers={"Authorization": f"Bearer {self.session.access_jwt}"}
        )
        return response.status_code == 200
    
    def delete_threadgate(self, rkey: str) -> None:
        """Delete a threadgate."""
        response = requests.post(
            f"{self.base_url}/com.atproto.repo.deleteRecord",
            json={
                "repo": self.session.did,
                "collection": "app.bsky.feed.threadgate",
                "rkey": rkey
            },
            headers={"Authorization": f"Bearer {self.session.access_jwt}"}
        )
        response.raise_for_status()

def create_session(identifier: str, password: str) -> BlueskySession:
    """Create a new Bluesky session."""
    response = requests.post(
        "https://bsky.social/xrpc/com.atproto.server.createSession",
        json={
            "identifier": identifier,
            "password": password
        }
    )
    response.raise_for_status()
    data = response.json()
    return BlueskySession(
        access_jwt=data["accessJwt"],
        did=data["did"]
    )

def extract_rkey(uri: str) -> str:
    """Extract the rkey from a threadgate or post URI."""
    return uri.split("/")[-1]

async def process_threadgates(client: BlueskyClient) -> tuple[int, int]:
    """Process all threadgates and return counts of checked and deleted."""
    checked = 0
    deleted = 0
    cursor = None
    
    with Progress(
        SpinnerColumn(),
        TextColumn("[progress.description]{task.description}"),
        console=console,
    ) as progress:
        task = progress.add_task("Processing threadgates...", total=None)
        
        while True:
            try:
                page = client.list_threadgates(cursor)
            except requests.RequestException as e:
                console.print(f"[red]Error fetching threadgates: {e}[/red]")
                break
                
            if not page.get("records"):
                break
                
            for record in page["records"]:
                checked += 1
                progress.update(task, description=f"Checking threadgate {checked}...")
                
                try:
                    post_rkey = extract_rkey(record["value"]["post"])
                    if not client.check_post_exists(post_rkey):
                        threadgate_rkey = extract_rkey(record["uri"])
                        client.delete_threadgate(threadgate_rkey)
                        deleted += 1
                        console.print(f"[yellow]Deleted abandoned threadgate {threadgate_rkey}[/yellow]")
                except requests.RequestException as e:
                    console.print(f"[red]Error processing threadgate: {e}[/red]")
                    continue
                    
                # Be kind to the API
                await asyncio.sleep(0.5)
            
            cursor = page.get("cursor")
            if not cursor:
                break
    
    return checked, deleted

@click.command()
def main():
    """Clean up abandoned threadgates on your Bluesky account."""
    console.print("[blue]🧹 Bluesky Abandoned Threadgate Cleaner[/blue]")
    
    try:
        identifier = Prompt.ask("Enter your Bluesky handle (e.g. username.bsky.social)")
        password = Prompt.ask("Enter your app-specific password", password=True)
        
        with console.status("[bold blue]Logging in...[/bold blue]"):
            session = create_session(identifier, password)
        console.print("[green]✓ Successfully logged in![/green]")
        
        client = BlueskyClient(session)
        checked, deleted = asyncio.run(process_threadgates(client))
        
        console.print("\n[bold]Summary:[/bold]")
        console.print(f"✓ Checked {checked} threadgates")
        console.print(f"🗑 Deleted {deleted} abandoned threadgates")
        
    except requests.RequestException as e:
        console.print(f"[red]Error: {e}[/red]")
        raise click.Abort()
    except KeyboardInterrupt:
        console.print("\n[yellow]Operation cancelled by user[/yellow]")
        raise click.Abort()

if __name__ == "__main__":
    main()

003-claude-v2.py

# /// script
# requires-python = ">=3.13"
# dependencies = [
#   "requests",
#   "rich"
# ]
# ///


import requests
from rich.console import Console
from rich.progress import Progress
from rich.table import Table
from rich import box
import getpass
import datetime

# Configure rich console
console = Console()
BASE_URL = "https://bsky.social/xrpc/"

def create_session(username, password):
    """Create a Bluesky session and return access token + DID"""
    url = BASE_URL + "com.atproto.server.createSession"
    response = requests.post(url, json={"identifier": username, "password": password})
    response.raise_for_status()
    data = response.json()
    return data["accessJwt"], data["did"]

def list_threadgates(did, access_jwt):
    """List all threadgates with pagination"""
    headers = {"Authorization": f"Bearer {access_jwt}"}
    records = []
    cursor = None

    with console.status("[bold cyan]Fetching threadgates...[/]", spinner="dots"):
        while True:
            params = {"repo": did, "collection": "app.bsky.feed.threadgate", "limit": 100}
            if cursor:
                params["cursor"] = cursor

            response = requests.get(
                BASE_URL + "com.atproto.repo.listRecords",
                params=params,
                headers=headers
            )
            response.raise_for_status()
            data = response.json()
            records.extend(data.get("records", []))
            cursor = data.get("cursor")
            if not cursor:
                break

    return records

def post_exists(did, post_rkey, access_jwt):
    """Check if a post exists"""
    headers = {"Authorization": f"Bearer {access_jwt}"}
    params = {"repo": did, "collection": "app.bsky.feed.post", "rkey": post_rkey}

    response = requests.get(
        BASE_URL + "com.atproto.repo.getRecord",
        params=params,
        headers=headers
    )
    return response.status_code == 200

def delete_threadgate(did, rkey, access_jwt):
    """Delete a threadgate record"""
    headers = {"Authorization": f"Bearer {access_jwt}"}
    data = {"repo": did, "collection": "app.bsky.feed.threadgate", "rkey": rkey}

    response = requests.post(
        BASE_URL + "com.atproto.repo.deleteRecord",
        json=data,
        headers=headers
    )
    response.raise_for_status()

def is_threadgate_incorrect(threadgate_value):
    """Check if a threadgate has incorrect allow settings"""
    allow = threadgate_value.get("allow")
    
    # Incorrect if no "allow" field at all
    if allow is None:
        return True
    
    # Incorrect if "allow" is not an empty array
    if allow != []:
        return True
    
    return False

def update_threadgate(did, rkey, post_uri, access_jwt):
    """Update a threadgate to have the correct empty allow array"""
    headers = {"Authorization": f"Bearer {access_jwt}"}
    
    # Create the corrected threadgate value
    threadgate_value = {
        "post": post_uri,
        "$type": "app.bsky.feed.threadgate",
        "allow": [],  # This is the correct setting
        "createdAt": datetime.datetime.now(datetime.UTC).isoformat().replace('+00:00', 'Z')
    }
    
    data = {
        "repo": did,
        "collection": "app.bsky.feed.threadgate",
        "rkey": rkey,
        "record": threadgate_value
    }

    response = requests.post(
        BASE_URL + "com.atproto.repo.putRecord",
        json=data,
        headers=headers
    )
    response.raise_for_status()

def main():
    console.print("\n[bold magenta]✨ Bluesky Threadgate Cleanup & Fix Tool ✨[/bold magenta]\n")

    # Get credentials
    username = input("Enter your username (e.g., user.bsky.social): ")
    password = getpass.getpass("Enter your app-specific password: ")

    try:
        # Create session
        with console.status("[bold green]Authenticating...[/]"):
            access_jwt, did = create_session(username, password)
        console.print(f"✅ [bold green]Authenticated as [cyan]{did}[/cyan][/bold green]\n")

        # Get all threadgates
        threadgates = list_threadgates(did, access_jwt)
        if not threadgates:
            console.print("[bold yellow]🎉 No threadgates found! Nothing to process.[/]")
            return

        console.print(f"🔍 Found [bold cyan]{len(threadgates)}[/] threadgates to check\n")

        # Process threadgates
        deleted_count = 0
        fixed_count = 0
        kept_count = 0
        
        with Progress() as progress:
            task = progress.add_task("[cyan]Processing threadgates...", total=len(threadgates))

            for record in threadgates:
                try:
                    # Extract keys
                    tg_uri = record["uri"]
                    tg_rkey = tg_uri.split("/")[-1]
                    post_uri = record["value"]["post"]
                    post_rkey = post_uri.split("/")[-1]

                    # Check if post still exists
                    if not post_exists(did, post_rkey, access_jwt):
                        console.print(f"  🗑️ [yellow]Deleting abandoned threadgate [bold]{tg_rkey}[/][/]")
                        delete_threadgate(did, tg_rkey, access_jwt)
                        deleted_count += 1
                        progress.console.print(f"  ✅ [red]Deleted abandoned [bold]{tg_rkey}[/][/]")
                    
                    # Check if threadgate has incorrect settings
                    elif is_threadgate_incorrect(record["value"]):
                        allow_value = record["value"].get("allow", "missing")
                        console.print(f"  🔧 [yellow]Fixing incorrect threadgate [bold]{tg_rkey}[/] (allow: {allow_value})[/]")
                        update_threadgate(did, tg_rkey, post_uri, access_jwt)
                        fixed_count += 1
                        progress.console.print(f"  ✅ [green]Fixed [bold]{tg_rkey}[/][/]")
                    
                    # Threadgate is already correct
                    else:
                        kept_count += 1
                        progress.console.print(f"  🔗 [grey58]Keeping correct threadgate [bold]{tg_rkey}[/][/]")

                except Exception as e:
                    console.print(f"[red]⚠️ Error processing {tg_rkey}: {str(e)}[/]")
                finally:
                    progress.update(task, advance=1)

        # Show summary
        console.print("\n[bold magenta]🏁 Processing Complete![/]")
        table = Table(box=box.ROUNDED, show_header=False, padding=(0, 2))
        table.add_row("Total Threadgates Checked", f"[bold]{len(threadgates)}[/]")
        table.add_row("Abandoned Threadgates Deleted", f"[bold red]{deleted_count}[/]")
        table.add_row("Incorrect Threadgates Fixed", f"[bold yellow]{fixed_count}[/]")
        table.add_row("Correct Threadgates Kept", f"[bold green]{kept_count}[/]")
        console.print(table)

        if fixed_count > 0:
            console.print(f"\n✨ [bold green]Fixed {fixed_count} threadgates to disable all replies![/]")

    except Exception as e:
        console.print(f"[bold red]🚨 Critical Error: {str(e)}[/]")
        raise

if __name__ == "__main__":
    main()

004-claude-4.6-opus.py

#!/usr/bin/env python3
# /// script
# requires-python = ">=3.13"
# dependencies = [
#   "requests",
#   "rich"
# ]
# ///
"""
🧹 Threadgate Janitor
Finds and removes abandoned threadgates — reply restrictions left behind
after their parent post was deleted.
"""

import sys
import time
import getpass
from dataclasses import dataclass, field

import requests
from rich.console import Console
from rich.live import Live
from rich.panel import Panel
from rich.progress import Progress, SpinnerColumn, TextColumn, BarColumn, MofNCompleteColumn
from rich.table import Table
from rich.text import Text
from rich import box

console = Console()

BASE = "https://bsky.social/xrpc"
# Be a polite citizen: small pause between API calls
REQUEST_DELAY = 0.05


@dataclass
class Stats:
    threadgates_found: int = 0
    posts_alive: int = 0
    posts_missing: int = 0
    deleted: int = 0
    errors: int = 0
    error_details: list = field(default_factory=list)


def authenticate(session: requests.Session, handle: str, password: str) -> tuple[str, str]:
    """Authenticate and return (access_jwt, did)."""
    resp = session.post(f"{BASE}/com.atproto.server.createSession", json={
        "identifier": handle,
        "password": password,
    })
    resp.raise_for_status()
    data = resp.json()
    return data["accessJwt"], data["did"]


def fetch_all_threadgates(session: requests.Session, did: str) -> list[dict]:
    """Paginate through all threadgate records."""
    records = []
    cursor = None

    with Progress(
        SpinnerColumn("dots"),
        TextColumn("[bold cyan]Fetching threadgates…"),
        BarColumn(bar_width=30),
        MofNCompleteColumn(),
        console=console,
        transient=True,
    ) as progress:
        task = progress.add_task("fetch", total=None)

        while True:
            params = {
                "repo": did,
                "collection": "app.bsky.feed.threadgate",
                "limit": 100,
            }
            if cursor:
                params["cursor"] = cursor

            resp = session.get(f"{BASE}/com.atproto.repo.listRecords", params=params)
            resp.raise_for_status()
            data = resp.json()

            batch = data.get("records", [])
            records.extend(batch)
            progress.update(task, completed=len(records))

            cursor = data.get("cursor")
            if not cursor or not batch:
                break

            time.sleep(REQUEST_DELAY)

    return records


def rkey_from_uri(uri: str) -> str:
    """Extract the record key (last segment) from an at:// URI."""
    return uri.rsplit("/", 1)[-1]


def post_exists(session: requests.Session, did: str, rkey: str, token: str) -> bool:
    """Check whether a post record still exists. Returns True if alive."""
    resp = session.get(f"{BASE}/com.atproto.repo.getRecord", params={
        "repo": did,
        "collection": "app.bsky.feed.post",
        "rkey": rkey,
    }, headers={"Authorization": f"Bearer {token}"})
    return resp.status_code == 200


def delete_threadgate(session: requests.Session, did: str, rkey: str, token: str) -> bool:
    """Delete a threadgate record. Returns True on success."""
    resp = session.post(f"{BASE}/com.atproto.repo.deleteRecord", json={
        "repo": did,
        "collection": "app.bsky.feed.threadgate",
        "rkey": rkey,
    }, headers={"Authorization": f"Bearer {token}"})
    return resp.status_code == 200


def build_summary(stats: Stats, elapsed: float) -> Panel:
    """Build a rich summary panel."""
    table = Table(show_header=False, box=None, padding=(0, 2))
    table.add_column(style="dim")
    table.add_column(style="bold")

    table.add_row("Threadgates scanned", str(stats.threadgates_found))
    table.add_row("Posts still alive", f"[green]{stats.posts_alive}[/]")
    table.add_row("Orphaned & deleted", f"[red]{stats.deleted}[/]")
    if stats.errors:
        table.add_row("Errors", f"[yellow]{stats.errors}[/]")
    table.add_row("Time elapsed", f"{elapsed:.1f}s")

    return Panel(table, title="[bold]Summary[/]", border_style="bright_blue", box=box.ROUNDED)


def main():
    console.print()
    console.print(
        Panel(
            "[bold]Threadgate Janitor[/]\n"
            "[dim]Sweeps away reply restrictions orphaned by deleted posts[/]",
            box=box.DOUBLE,
            border_style="bright_magenta",
            padding=(1, 4),
        )
    )
    console.print()

    # --- Credentials ---
    handle = console.input("[bold cyan]Handle:[/] ")
    password = getpass.getpass("App password: ")
    console.print()

    # --- Session setup ---
    session = requests.Session()
    session.headers["Accept"] = "application/json"

    # --- Auth ---
    with console.status("[bold cyan]Authenticating…[/]", spinner="dots"):
        try:
            token, did = authenticate(session, handle, password)
        except requests.HTTPError as e:
            console.print(f"[bold red]Authentication failed:[/] {e.response.status_code} — double-check your handle and app password.")
            sys.exit(1)

    console.print(f"  [green]✓[/] Logged in as [bold]{handle}[/]  [dim]({did})[/]")
    console.print()

    # --- Fetch threadgates ---
    records = fetch_all_threadgates(session, did)

    if not records:
        console.print("  [green]✓[/] No threadgates found — nothing to clean up!")
        return

    console.print(f"  [cyan]Found {len(records)} threadgate(s) to inspect.[/]")
    console.print()

    # --- Inspect & clean ---
    stats = Stats(threadgates_found=len(records))
    start = time.monotonic()

    with Progress(
        SpinnerColumn("dots"),
        TextColumn("[bold]{task.description}"),
        BarColumn(bar_width=30),
        MofNCompleteColumn(),
        TextColumn("[dim]{task.fields[status]}[/]"),
        console=console,
    ) as progress:
        task = progress.add_task("Inspecting", total=len(records), status="")

        for rec in records:
            tg_uri = rec["uri"]
            tg_rkey = rkey_from_uri(tg_uri)
            post_uri = rec.get("value", {}).get("post", "")
            post_rkey = rkey_from_uri(post_uri) if post_uri else None

            short_id = tg_rkey[:12]

            if not post_rkey:
                # Malformed threadgate with no post reference — treat as orphan
                stats.posts_missing += 1
                progress.update(task, status=f"[yellow]⚠ {short_id} — no post ref[/]")
            elif post_exists(session, did, post_rkey, token):
                stats.posts_alive += 1
                progress.update(task, status=f"[green]✓ {short_id}[/]")
                progress.advance(task)
                time.sleep(REQUEST_DELAY)
                continue
            else:
                stats.posts_missing += 1
                progress.update(task, status=f"[red]✗ {short_id} — orphaned[/]")

            # Delete the orphan
            time.sleep(REQUEST_DELAY)
            if delete_threadgate(session, did, tg_rkey, token):
                stats.deleted += 1
                progress.update(task, status=f"[red]🗑 {short_id} — removed[/]")
            else:
                stats.errors += 1
                stats.error_details.append(tg_rkey)
                progress.update(task, status=f"[yellow]⚠ {short_id} — delete failed[/]")

            progress.advance(task)
            time.sleep(REQUEST_DELAY)

    elapsed = time.monotonic() - start
    console.print()
    console.print(build_summary(stats, elapsed))

    if stats.errors:
        console.print(f"\n  [yellow]Could not delete:[/] {', '.join(stats.error_details)}")

    if stats.deleted:
        console.print(f"\n  [bright_magenta]Swept away {stats.deleted} orphan(s). Tidy! 🧹[/]")
    else:
        console.print(f"\n  [green]All threadgates are accounted for — nothing to clean.[/]")

    console.print()


if __name__ == "__main__":
    main()