absurdctl

#!/usr/bin/env python3
import csv
import io
import json
import os
import re
import subprocess
import sys
import shutil
import textwrap
from urllib.request import urlopen
from urllib.error import URLError
from urllib.parse import quote, urlparse
from optparse import OptionParser, IndentedHelpFormatter


# Queue names are interpolated into generated table/index names (t_<queue>,
# r_<queue>_sai, ...). We cap UTF-8 byte length so those identifiers stay
# within PostgreSQL's 63-byte limit; character set is otherwise delegated to
# PostgreSQL quoted identifier rules.
MAX_QUEUE_NAME_LENGTH = 57


GITHUB_OWNER = "earendil-works"
GITHUB_REPO = "absurd"
GITHUB_RAW_BASE = f"https://raw.githubusercontent.com/{GITHUB_OWNER}/{GITHUB_REPO}"
GITHUB_API_BASE = f"https://api.github.com/repos/{GITHUB_OWNER}/{GITHUB_REPO}"

# BEGIN BUNDLED DATA (autogenerated by scripts/build-absurdctl)
ABSURD_SCHEMA_TARGET_VERSION = "main"
BUNDLED_SCHEMA_SQL = None
BUNDLED_MIGRATIONS = {}
BUNDLED_SKILLS = {}
# END BUNDLED DATA

MIGRATION_FILENAME_RE = re.compile(
    r"^(?P<from>[0-9]+\.[0-9]+\.[0-9]+(?:-[A-Za-z0-9.]+)?)-(?P<to>main|[0-9]+\.[0-9]+\.[0-9]+(?:-[A-Za-z0-9.]+)?)\.sql$"
)
VERSION_RE = re.compile(
    r"^(?P<major>[0-9]+)\.(?P<minor>[0-9]+)\.(?P<patch>[0-9]+)(?:-(?P<suffix>[A-Za-z0-9.]+))?$"
)
SCHEMA_VERSION_FUNCTION_RE = re.compile(
    r"create\s+or\s+replace\s+function\s+absurd\.get_schema_version\s*\(",
    re.IGNORECASE,
)


def quote_ident(identifier):
    """Quote a SQL identifier using PostgreSQL double-quote escaping."""
    return '"' + identifier.replace('"', '""') + '"'


def queue_relation(prefix, queue_name):
    """Return a fully qualified absurd.<prefix>_<queue> relation reference."""
    return f"absurd.{quote_ident(f'{prefix}_{queue_name}')}"


def validate_queue_name(value):
    """Validate queue names used in dynamic table/index names."""
    if value is None or value == "":
        print("Error: Queue name must be provided.", file=sys.stderr)
        sys.exit(1)

    if len(value.encode("utf-8")) > MAX_QUEUE_NAME_LENGTH:
        print(
            f"Error: Queue name '{value}' is too long (max {MAX_QUEUE_NAME_LENGTH} bytes).",
            file=sys.stderr,
        )
        sys.exit(1)

    return value


def normalize_partition_table_name(value):
    """Normalize a partition table reference to an absurd-schema table name."""
    if value is None:
        print("Error: Partition table must be provided.", file=sys.stderr)
        sys.exit(1)

    partition_table = value.strip()
    if not partition_table:
        print("Error: Partition table must be provided.", file=sys.stderr)
        sys.exit(1)

    lower_partition_table = partition_table.lower()
    if lower_partition_table.startswith("absurd."):
        partition_table = partition_table[len("absurd.") :]
    elif lower_partition_table.startswith('"absurd".'):
        partition_table = partition_table[len('"absurd".') :]

    if partition_table.startswith('"') and partition_table.endswith('"'):
        partition_table = partition_table[1:-1].replace('""', '"')

    if not partition_table:
        print("Error: Partition table must be provided.", file=sys.stderr)
        sys.exit(1)

    return partition_table


def build_drop_partition_sql(partition_table):
    return f"drop table if exists absurd.{quote_ident(partition_table)}"


def build_detach_partition_sql(config, parent_table, partition_table):
    has_default = run_psql(
        config,
        """
        select exists (
          select 1
          from pg_class parent
          join pg_namespace pn on pn.oid = parent.relnamespace
          join pg_inherits inh on inh.inhparent = parent.oid
          join pg_class child on child.oid = inh.inhrelid
          where pn.nspname = 'absurd'
            and parent.relname = :'parent_table'
            and pg_get_expr(child.relpartbound, child.oid) = 'DEFAULT'
        )
        """,
        tuples_only=True,
        no_align=True,
        variables={"parent_table": parent_table},
    ).strip()

    sql = (
        f"alter table absurd.{quote_ident(parent_table)} "
        f"detach partition absurd.{quote_ident(partition_table)}"
    )
    if has_default != "t":
        sql += " concurrently"
    return sql


class RemoteMigrationDiscoveryError(Exception):
    pass


def get_term_width():
    """Get terminal width for formatting."""
    try:
        return shutil.get_terminal_size().columns
    except Exception:
        return 80


class CleanHelpFormatter(IndentedHelpFormatter):
    """Custom formatter with better spacing and formatting."""

    def __init__(self):
        IndentedHelpFormatter.__init__(
            self,
            indent_increment=2,
            max_help_position=30,
            width=get_term_width(),
            short_first=1,
        )

    def format_option(self, option):
        result = []
        opts = self.option_strings[option]
        opt_width = self.help_position - self.current_indent - 2

        if len(opts) > opt_width:
            opts = "%*s%s\n" % (self.current_indent, "", opts)
            indent_first = self.help_position
        else:
            opts = "%*s%-*s  " % (self.current_indent, "", opt_width, opts)
            indent_first = 0

        result.append(opts)

        if option.help:
            help_text = self.expand_default(option)
            help_lines = []

            help_lines = textwrap.wrap(help_text, self.width - self.help_position)

            result.append("%*s%s\n" % (indent_first, "", help_lines[0]))
            for line in help_lines[1:]:
                result.append("%*s%s\n" % (self.help_position, "", line))
        elif opts[-1] != "\n":
            result.append("\n")

        return "".join(result)


def build_database_config(host, port, user, database):
    """Build a dict with database configuration."""
    return {
        "host": host,
        "port": port,
        "user": user,
        "database": database,
    }


def resolve_database_argument(explicit_database=None):
    """Resolve database/URI using explicit args first, then env vars, then fallback."""
    return (
        explicit_database
        or os.getenv("ABSURD_DATABASE_URL")
        or os.getenv("PGDATABASE")
        or "postgresql://localhost/absurd"
    )


def config_from_options(options):
    """Shortcut to construct a database config from parsed options."""
    database = resolve_database_argument(options.database)
    parsed = urlparse(database)
    if parsed.scheme and "postgres" in parsed.scheme:
        return {"uri": database}

    return build_database_config(
        options.host or os.getenv("PGHOST", "localhost"),
        options.port or os.getenv("PGPORT", "5432"),
        options.user or os.getenv("PGUSER", os.getenv("USER", "")),
        database,
    )


def build_psql_command(config, extra_args=None):
    """Construct the base psql command with optional additional arguments."""
    cmd = ["psql", "-X"]

    if "uri" in config:
        cmd.extend(["-d", config["uri"]])
    else:
        cmd.extend(
            [
                "-h",
                config["host"],
                "-p",
                config["port"],
                "-U",
                config["user"],
                "-d",
                config["database"],
            ]
        )

    if extra_args:
        cmd.extend(extra_args)

    return cmd


def sanitize_uri(uri):
    """Remove password from URI for safe display in verbose output."""
    parsed = urlparse(uri)
    if parsed.password:
        return uri.replace(f"{parsed.password}@", f"****@", 1)
    return uri


def get_common_db_details(config):
    """Return standard database connection details for verbose output."""
    if "uri" in config:
        sanitized_uri = sanitize_uri(config["uri"])
        return [("URI", sanitized_uri)]
    return [
        ("Host", config["host"]),
        ("Port", config["port"]),
        ("User", config["user"]),
        ("Database", config["database"]),
    ]


def print_verbose_configuration(enabled, details):
    """Print a formatted configuration block when verbose output is enabled."""
    if not enabled:
        return
    print("Configuration:")
    for label, value in details:
        print(f"  {label}: {value}")


def run_psql(
    config,
    query=None,
    *,
    tuples_only=False,
    no_align=False,
    extra_args=None,
    input_data=None,
    error_message=None,
    strip_output=True,
    variables=None,
    fail_on_error=True,
):
    """
    Execute a PostgreSQL command. Returns stripped stdout for queries and the
    CompletedProcess for other invocations.

    Use `variables` dict to pass values safely. In the query, reference them as :'name'
    for string values. Example:
        run_psql(config, "SELECT * FROM t WHERE id = :'id'", variables={"id": user_input})

    Note: psql does not expand :variables for SQL passed via -c, so SQL text is
    sent through stdin with -f - whenever query/input_data is provided.
    """
    cmd_args = []
    stdin_data = input_data

    if tuples_only:
        cmd_args.append("-t")
    if no_align:
        cmd_args.append("-A")
    if fail_on_error:
        cmd_args.extend(["-v", "ON_ERROR_STOP=1"])
    if variables:
        for name, value in variables.items():
            cmd_args.extend(["-v", f"{name}={value}"])
    if extra_args:
        cmd_args.extend(extra_args)
    if query is not None:
        if input_data is not None:
            raise ValueError("run_psql received both query and input_data")
        stdin_data = query

    if stdin_data is not None:
        cmd_args.extend(["-f", "-"])

    message = error_message
    if message is None:
        message = (
            "Error executing query"
            if query is not None
            else "Error executing psql command"
        )

    env = os.environ.copy()
    env["PGPASSWORD"] = os.getenv("PGPASSWORD", "")
    try:
        result = subprocess.run(
            build_psql_command(config, cmd_args),
            input=stdin_data,
            capture_output=True,
            text=True,
            check=True,
            env=env,
        )
    except subprocess.CalledProcessError as e:
        if fail_on_error:
            print(f"{message}: {e.stderr}", file=sys.stderr)
            sys.exit(1)
        raise

    if query is not None:
        return result.stdout.strip() if strip_output else result.stdout
    return result


def run_psql_csv(config, query, variables=None):
    """Execute a query and return rows parsed via CSV output."""
    output = run_psql(
        config,
        query,
        tuples_only=True,
        extra_args=["--csv"],
        strip_output=False,
        variables=variables,
    )
    if not output:
        return []

    reader = csv.reader(io.StringIO(output))
    return [row for row in reader if row]


def queue_exists(config, queue_name):
    """Check if a queue exists."""
    result = run_psql(
        config,
        "SELECT COUNT(*) FROM absurd.queues WHERE queue_name = :'queue_name';",
        tuples_only=True,
        no_align=True,
        variables={"queue_name": queue_name},
    )
    return int(result) > 0


def build_json_params(param_list, base_params=None):
    """
    Build JSON parameters from -p flags and optional base --params.

    Supports:
      -p foo=bar           -> {"foo": "bar"}
      -p count:=42         -> {"count": 42}
      -p enabled:=true     -> {"enabled": true}
      -p tags:='["a","b"]' -> {"tags": ["a","b"]}
      -p user.name=Alice   -> {"user": {"name": "Alice"}}
      --params '{"x":1}'   -> merged with -p flags
    """
    result = {}

    if base_params:
        try:
            result = json.loads(base_params)
            if not isinstance(result, dict):
                raise ValueError("--params must be a JSON object")
        except json.JSONDecodeError as e:
            print(f"Error parsing --params: {e}", file=sys.stderr)
            sys.exit(1)

    for param in param_list:
        if ":=" in param:
            key, value_str = param.split(":=", 1)
            try:
                value = json.loads(value_str)
            except json.JSONDecodeError as e:
                print(f"Error parsing JSON value in '{param}': {e}", file=sys.stderr)
                sys.exit(1)
        elif "=" in param:
            key, value = param.split("=", 1)
        else:
            print(
                f"Invalid parameter format: '{param}' (expected key=value or key:=json)",
                file=sys.stderr,
            )
            sys.exit(1)

        set_nested_value(result, key, value)

    return json.dumps(result)


def set_nested_value(obj, path, value):
    """Set a value in a nested dictionary using dot notation."""
    keys = path.split(".")
    current = obj

    for key in keys[:-1]:
        if key not in current:
            current[key] = {}
        elif not isinstance(current[key], dict):
            print(
                f"Error: Cannot set nested value, '{key}' is not an object",
                file=sys.stderr,
            )
            sys.exit(1)
        current = current[key]

    current[keys[-1]] = value


def add_db_options(parser):
    """Add common database connection options to a parser."""
    parser.add_option("-h", "--host", dest="host", metavar="HOST", help="Database host")
    parser.add_option("-p", "--port", dest="port", metavar="PORT", help="Database port")
    parser.add_option("-U", "--user", dest="user", metavar="USER", help="Database user")
    parser.add_option(
        "-d",
        "--database",
        dest="database",
        metavar="DB",
        help="Database name or PostgreSQL connection URI",
    )
    parser.add_option(
        "-v",
        "--verbose",
        dest="verbose",
        action="store_true",
        default=False,
        help="Verbose output",
    )


def build_command_parser(usage):
    """Create an OptionParser with common Absurd defaults."""
    parser = OptionParser(
        usage=usage,
        add_help_option=False,
        formatter=CleanHelpFormatter(),
    )
    add_db_options(parser)
    parser.add_option("--help", action="store_true", help="Show this help message")
    return parser


def parse_args_with_help(parser, args, examples=None):
    """Parse args and handle --help in a consistent way."""
    options, remaining = parser.parse_args(args)
    if getattr(options, "help", False):
        parser.print_help()
        if examples:
            print("\nExamples:")
            for example in examples:
                print(f"  {example}")
        sys.exit(0)
    return options, remaining


def ensure_queue_exists(config, queue_name):
    """Exit with error if queue is missing."""
    if not queue_exists(config, queue_name):
        print(f"Queue '{queue_name}' does not exist", file=sys.stderr)
        sys.exit(1)


def github_absurd_sql_url(ref):
    return f"{GITHUB_RAW_BASE}/{quote(ref)}/sql/absurd.sql"


def github_migration_index_url(ref):
    return f"{GITHUB_API_BASE}/contents/sql/migrations?ref={quote(ref)}"


def github_migration_sql_url(ref, migration_name):
    return f"{GITHUB_RAW_BASE}/{quote(ref)}/sql/migrations/{quote(migration_name)}"


def github_repo_file_url(ref, relative_path):
    return f"{GITHUB_RAW_BASE}/{quote(ref)}/{quote(relative_path, safe='/')}"


def fetch_text(url):
    with urlopen(url) as response:
        return response.read().decode("utf-8")


def parse_migration_filename(filename):
    match = MIGRATION_FILENAME_RE.match(filename)
    if not match:
        return None
    return match.group("from"), match.group("to")


def version_sort_key(version):
    if version == "main":
        return (10**9, 10**9, 10**9, 1, "")

    match = VERSION_RE.match(version)
    if match is None:
        # Keep unknown version strings stable and after regular semver tags.
        return (10**8, 0, 0, 0, version)

    suffix = match.group("suffix") or ""
    # Stable release sorts after pre-release of same x.y.z.
    stable_flag = 1 if suffix == "" else 0
    return (
        int(match.group("major")),
        int(match.group("minor")),
        int(match.group("patch")),
        stable_flag,
        suffix,
    )


def compare_versions(a, b):
    ka = version_sort_key(a)
    kb = version_sort_key(b)
    if ka < kb:
        return -1
    if ka > kb:
        return 1
    return 0


def migration_declares_schema_version(sql_content):
    return SCHEMA_VERSION_FUNCTION_RE.search(sql_content) is not None


def get_recorded_schema_version(config):
    exists = run_psql(
        config,
        "SELECT to_regprocedure('absurd.get_schema_version()') IS NOT NULL;",
        tuples_only=True,
        no_align=True,
        error_message="Error checking schema version metadata",
    )
    if exists.lower() not in ("t", "true", "1"):
        return None

    version = run_psql(
        config,
        "SELECT absurd.get_schema_version();",
        tuples_only=True,
        no_align=True,
        error_message="Error loading schema version",
    )
    return version or None


def load_absurd_sql(script_dir, target_ref, verbose):
    local_sql_path = os.path.join(script_dir, "sql", "absurd.sql")

    if os.path.exists(local_sql_path) and target_ref == ABSURD_SCHEMA_TARGET_VERSION:
        if verbose:
            print(f"Using local SQL file: {local_sql_path}")
        with open(local_sql_path, "r", encoding="utf-8") as f:
            return f.read(), f"local:{local_sql_path}"

    if BUNDLED_SCHEMA_SQL and target_ref == ABSURD_SCHEMA_TARGET_VERSION:
        if verbose:
            print("Using bundled absurd.sql")
        return BUNDLED_SCHEMA_SQL, "bundled"

    sql_url = github_absurd_sql_url(target_ref)
    if verbose:
        print(f"Fetching SQL file from: {sql_url}")
    try:
        return fetch_text(sql_url), f"github:{sql_url}"
    except URLError as e:
        print(f"Error downloading SQL file from GitHub: {e}", file=sys.stderr)
        sys.exit(1)


def discover_local_migrations(script_dir):
    migrations = {}
    migrations_dir = os.path.join(script_dir, "sql", "migrations")
    if not os.path.isdir(migrations_dir):
        return migrations

    for filename in sorted(os.listdir(migrations_dir)):
        parsed = parse_migration_filename(filename)
        if parsed is None:
            continue
        source, target = parsed
        path = os.path.join(migrations_dir, filename)
        with open(path, "r", encoding="utf-8") as f:
            migrations[(source, target)] = {
                "name": filename,
                "source": "local",
                "sql": f.read(),
            }
    return migrations


def discover_bundled_migrations():
    migrations = {}
    for filename, sql_content in BUNDLED_MIGRATIONS.items():
        parsed = parse_migration_filename(filename)
        if parsed is None:
            continue
        source, target = parsed
        migrations[(source, target)] = {
            "name": filename,
            "source": "bundled",
            "sql": sql_content,
        }
    return migrations


def discover_remote_migrations(target_ref):
    migrations = {}
    index_url = github_migration_index_url(target_ref)

    try:
        payload = fetch_text(index_url)
    except URLError as e:
        raise RemoteMigrationDiscoveryError(
            f"failed to download migration index from {index_url}: {e}"
        ) from e

    try:
        entries = json.loads(payload)
    except json.JSONDecodeError as e:
        raise RemoteMigrationDiscoveryError(
            f"failed to parse migration index from {index_url}: {e}"
        ) from e

    if not isinstance(entries, list):
        raise RemoteMigrationDiscoveryError(
            f"unexpected migration index payload from {index_url}: expected JSON list"
        )

    for entry in entries:
        name = entry.get("name", "")
        parsed = parse_migration_filename(name)
        if parsed is None:
            continue
        source, target = parsed
        migrations[(source, target)] = {
            "name": name,
            "source": "github",
            "url": github_migration_sql_url(target_ref, name),
        }
    return migrations


def discover_migrations(script_dir, target_ref, include_remote=False):
    # Keep source selection simple: use local/bundled assets first and only
    # switch to remote discovery when explicitly requested.
    if include_remote:
        return discover_remote_migrations(target_ref)

    local = discover_local_migrations(script_dir)
    if local:
        return local

    bundled = discover_bundled_migrations()
    if bundled:
        return bundled

    return {}


def resolve_migration_path(current_version, target_version, migrations):
    direction = compare_versions(current_version, target_version)
    if direction == 0:
        return []
    if direction > 0:
        # Downgrades are not supported.
        return None

    # Migrations are applied in release order by destination version.
    # Missing release-to-release files are treated as null migrations.
    grouped_by_target = {}
    for (src, dst), migration in migrations.items():
        if compare_versions(dst, current_version) <= 0:
            continue
        if compare_versions(dst, target_version) > 0:
            continue
        grouped_by_target.setdefault(dst, []).append((src, migration))

    ordered_targets = sorted(grouped_by_target.keys(), key=version_sort_key)

    cursor = current_version
    path = []
    for dst in ordered_targets:
        candidates = grouped_by_target[dst]

        exact = [item for item in candidates if compare_versions(item[0], cursor) == 0]
        if exact:
            chosen = exact[0][1]
        else:
            le_cursor = [item for item in candidates if compare_versions(item[0], cursor) < 0]
            if le_cursor:
                # Prefer the newest compatible source version.
                chosen = max(le_cursor, key=lambda item: version_sort_key(item[0]))[1]
            else:
                # No direct predecessor migration is known; assume null migrations
                # from cursor up to the first available source.
                chosen = min(candidates, key=lambda item: version_sort_key(item[0]))[1]

        path.append(chosen)
        cursor = dst

    return path


def get_migration_sql(migration):
    if "sql" in migration:
        return migration["sql"]
    try:
        return fetch_text(migration["url"])
    except URLError as e:
        print(
            f"Error downloading migration {migration['name']} from GitHub: {e}",
            file=sys.stderr,
        )
        sys.exit(1)


def load_skill_file(script_dir, relative_path, verbose):
    local_path = os.path.join(script_dir, *relative_path.split("/"))

    if os.path.exists(local_path):
        if verbose:
            print(f"Using local skill file: {local_path}")
        with open(local_path, "r", encoding="utf-8") as f:
            return f.read(), f"local:{local_path}"

    bundled_content = BUNDLED_SKILLS.get(relative_path)
    if bundled_content is not None:
        if verbose:
            print(f"Using bundled skill file: {relative_path}")
        return bundled_content, "bundled"

    url = github_repo_file_url(ABSURD_SCHEMA_TARGET_VERSION, relative_path)
    if verbose:
        print(f"Fetching skill file from: {url}")
    try:
        return fetch_text(url), f"github:{url}"
    except URLError as e:
        print(f"Error downloading skill file from GitHub: {e}", file=sys.stderr)
        sys.exit(1)


def render_combined_migration_sql(current_version, target_version, resolved_steps):
    parts = [
        "-- Absurd migration SQL bundle",
        f"-- Range: {current_version} -> {target_version}",
        "",
    ]

    for step in resolved_steps:
        migration = step["migration"]
        sql_content = step["sql"]
        parts.append(
            f"-- begin migration: {migration['name']} [{migration['source']}]"
        )
        if sql_content:
            parts.append(sql_content)
        parts.append(f"-- end migration: {migration['name']}")
        parts.append("")

    return "\n".join(parts)


def cmd_cleanup(args):
    """Clean up old completed, failed, or cancelled tasks and events."""
    usage = "usage: %prog cleanup [options] QUEUE_NAME TTL_DAYS"
    parser = build_command_parser(usage)
    parser.add_option(
        "-l",
        "--limit",
        dest="limit",
        type="int",
        default=1000,
        help="Batch size for deletions (default: 1000)",
    )
    parser.add_option(
        "-n",
        "--dry-run",
        dest="dry_run",
        action="store_true",
        default=False,
        help="Show what would be deleted without actually deleting",
    )
    examples = [
        "absurdctl cleanup myqueue 30",
        "absurdctl cleanup --dry-run -v myqueue 30",
    ]

    options, args = parse_args_with_help(parser, args, examples)

    if len(args) != 2:
        print(
            "Error: Missing required arguments QUEUE_NAME and TTL_DAYS", file=sys.stderr
        )
        parser.print_help()
        sys.exit(1)

    queue_name = validate_queue_name(args[0])
    try:
        ttl_days = int(args[1])
    except ValueError:
        print(
            f"Error: TTL_DAYS must be a positive integer, got: {args[1]}",
            file=sys.stderr,
        )
        sys.exit(1)

    config = config_from_options(options)
    ensure_queue_exists(config, queue_name)

    ttl_seconds = ttl_days * 24 * 60 * 60

    print_verbose_configuration(
        options.verbose,
        [
            ("Queue", queue_name),
            ("TTL", f"{ttl_days} days ({ttl_seconds} seconds)"),
        ]
        + get_common_db_details(config)
        + [
            ("Limit", options.limit),
            ("Dry run", options.dry_run),
        ],
    )

    if options.dry_run:
        print("DRY RUN MODE - No data will be deleted")

        t_rel = queue_relation("t", queue_name)
        r_rel = queue_relation("r", queue_name)
        e_rel = queue_relation("e", queue_name)

        task_count = run_psql(
            config,
            f"""SELECT COUNT(*) FROM {t_rel} t
                LEFT JOIN {r_rel} r ON r.run_id = t.last_attempt_run
                WHERE t.state IN ('completed', 'failed', 'cancelled')
                AND ((t.state = 'completed' AND r.completed_at < NOW() - make_interval(secs => :'ttl_seconds'::int))
                OR (t.state = 'failed' AND r.failed_at < NOW() - make_interval(secs => :'ttl_seconds'::int))
                OR (t.state = 'cancelled' AND t.cancelled_at < NOW() - make_interval(secs => :'ttl_seconds'::int)));""",
            tuples_only=True,
            no_align=True,
            variables={"ttl_seconds": ttl_seconds},
        )

        event_count = run_psql(
            config,
            f"SELECT COUNT(*) FROM {e_rel} WHERE emitted_at < NOW() - make_interval(secs => :'ttl_seconds'::int);",
            tuples_only=True,
            no_align=True,
            variables={"ttl_seconds": ttl_seconds},
        )

        print(f"Would delete {task_count} tasks and {event_count} events")
        return

    print(f"Starting cleanup for queue '{queue_name}' (TTL: {ttl_days} days)")

    total_tasks_deleted = 0
    total_events_deleted = 0
    iteration = 0

    while True:
        iteration += 1

        if options.verbose:
            print(f"Iteration {iteration}...")

        tasks_deleted = int(
            run_psql(
                config,
                "SELECT absurd.cleanup_tasks(:'queue_name', :'ttl_seconds'::int, :'limit'::int);",
                tuples_only=True,
                no_align=True,
                variables={
                    "queue_name": queue_name,
                    "ttl_seconds": ttl_seconds,
                    "limit": options.limit,
                },
            )
        )
        total_tasks_deleted += tasks_deleted

        if options.verbose and tasks_deleted > 0:
            print(f"  Deleted {tasks_deleted} tasks")

        events_deleted = int(
            run_psql(
                config,
                "SELECT absurd.cleanup_events(:'queue_name', :'ttl_seconds'::int, :'limit'::int);",
                tuples_only=True,
                no_align=True,
                variables={
                    "queue_name": queue_name,
                    "ttl_seconds": ttl_seconds,
                    "limit": options.limit,
                },
            )
        )
        total_events_deleted += events_deleted

        if options.verbose and events_deleted > 0:
            print(f"  Deleted {events_deleted} events")

        if tasks_deleted == 0 and events_deleted == 0:
            break

        if tasks_deleted > 0 or events_deleted > 0:
            print(f"Deleted {tasks_deleted} tasks, {events_deleted} events")

    print(
        f"Cleanup complete: {total_tasks_deleted} tasks, {total_events_deleted} events deleted in {iteration} iterations"
    )


def cmd_create_queue(args):
    """Create a new Absurd queue."""
    usage = "usage: %prog create-queue [options] QUEUE_NAME"
    parser = build_command_parser(usage)
    parser.add_option(
        "--storage-mode",
        dest="storage_mode",
        type="choice",
        choices=["unpartitioned", "partitioned"],
        default="unpartitioned",
        help="Queue storage mode (default: unpartitioned)",
    )

    options, args = parse_args_with_help(
        parser,
        args,
        [
            "absurdctl create-queue myqueue",
            "absurdctl create-queue jobs --storage-mode partitioned",
        ],
    )

    if len(args) != 1:
        print("Error: Missing required argument QUEUE_NAME", file=sys.stderr)
        parser.print_help()
        sys.exit(1)

    queue_name = validate_queue_name(args[0])
    config = config_from_options(options)

    print_verbose_configuration(
        options.verbose,
        [("Queue", queue_name), ("Storage mode", options.storage_mode)]
        + get_common_db_details(config),
    )
    if options.verbose:
        print(f"Creating queue '{queue_name}'...")

    if options.storage_mode == "unpartitioned":
        run_psql(
            config,
            "SELECT absurd.create_queue(:'queue_name');",
            variables={"queue_name": queue_name},
        )
    else:
        run_psql(
            config,
            "SELECT absurd.create_queue(:'queue_name', :'storage_mode');",
            variables={
                "queue_name": queue_name,
                "storage_mode": options.storage_mode,
            },
        )
    print(f"Queue '{queue_name}' created successfully")


def cmd_drop_queue(args):
    """Drop an existing Absurd queue."""
    usage = "usage: %prog drop-queue [options] QUEUE_NAME"
    parser = build_command_parser(usage)
    parser.add_option(
        "-y",
        "--yes",
        dest="yes",
        action="store_true",
        default=False,
        help="Skip confirmation prompt",
    )
    examples = [
        "absurdctl drop-queue myqueue",
        "absurdctl drop-queue --yes myqueue",
    ]

    options, args = parse_args_with_help(parser, args, examples)

    if len(args) != 1:
        print("Error: Missing required argument QUEUE_NAME", file=sys.stderr)
        parser.print_help()
        sys.exit(1)

    queue_name = validate_queue_name(args[0])
    config = config_from_options(options)

    ensure_queue_exists(config, queue_name)

    print_verbose_configuration(
        options.verbose,
        [("Queue", queue_name)] + get_common_db_details(config),
    )

    if not options.yes:
        print(
            f"WARNING: This will permanently delete queue '{queue_name}' and all its data."
        )
        response = input("Are you sure you want to continue? (yes/no): ")
        if response != "yes":
            print("Operation cancelled")
            return

    if options.verbose:
        print(f"Dropping queue '{queue_name}'...")

    run_psql(config, "SELECT absurd.drop_queue(:'queue_name');", variables={"queue_name": queue_name})
    print(f"Queue '{queue_name}' dropped successfully")


def cmd_list_queues(args):
    """List all existing Absurd queues."""
    usage = "usage: %prog list-queues [options]"
    parser = build_command_parser(usage)

    options, args = parse_args_with_help(parser, args)

    config = config_from_options(options)

    result = run_psql(
        config, "SELECT * FROM absurd.list_queues();", tuples_only=True, no_align=True
    ).strip()
    if result:
        print(result)


def _fetch_queue_policy(config, queue_name):
    rows = run_psql_csv(
        config,
        """
        SELECT
          queue_name,
          storage_mode,
          default_partition,
          partition_lookahead::text,
          partition_lookback::text,
          cleanup_ttl::text,
          cleanup_limit,
          detach_mode,
          detach_min_age::text
        FROM absurd.get_queue_policy(:'queue_name')
        """,
        variables={"queue_name": queue_name},
    )
    if not rows:
        return None

    parts = rows[0]
    if len(parts) < 9:
        return None

    return {
        "queue_name": parts[0],
        "storage_mode": parts[1],
        "default_partition": parts[2],
        "partition_lookahead": parts[3],
        "partition_lookback": parts[4],
        "cleanup_ttl": parts[5],
        "cleanup_limit": parts[6],
        "detach_mode": parts[7],
        "detach_min_age": parts[8],
    }


def _print_queue_policy(policy):
    print(f"Queue policy for '{policy['queue_name']}':")
    print(f"  storage_mode: {policy['storage_mode']}")
    print(f"  default_partition: {policy['default_partition']}")
    print(f"  partition_lookahead: {policy['partition_lookahead']}")
    print(f"  partition_lookback: {policy['partition_lookback']}")
    print(f"  cleanup_ttl: {policy['cleanup_ttl']}")
    print(f"  cleanup_limit: {policy['cleanup_limit']}")
    print(f"  detach_mode: {policy['detach_mode']}")
    print(f"  detach_min_age: {policy['detach_min_age']}")


def cmd_queue_policy(args):
    """Get or update queue maintenance policy."""
    usage = "usage: %prog queue-policy [options] QUEUE_NAME"
    parser = build_command_parser(usage)
    parser.add_option(
        "--default-partition",
        dest="default_partition",
        type="choice",
        choices=["enabled", "disabled"],
        metavar="MODE",
        help="Default partition mode for partitioned queues (enabled or disabled)",
    )
    parser.add_option(
        "--partition-lookahead",
        dest="partition_lookahead",
        metavar="INTERVAL",
        help="Policy partition lookahead interval (for example: '28 days')",
    )
    parser.add_option(
        "--partition-lookback",
        dest="partition_lookback",
        metavar="INTERVAL",
        help="Policy partition lookback interval (for example: '1 day')",
    )
    parser.add_option(
        "--cleanup-ttl",
        dest="cleanup_ttl",
        metavar="INTERVAL",
        help="Cleanup retention interval (for example: '30 days')",
    )
    parser.add_option(
        "--cleanup-limit",
        dest="cleanup_limit",
        type="int",
        metavar="N",
        help="Cleanup batch size",
    )
    parser.add_option(
        "--detach-mode",
        dest="detach_mode",
        type="choice",
        choices=["none", "empty"],
        metavar="MODE",
        help="Detach mode (none or empty)",
    )
    parser.add_option(
        "--detach-min-age",
        dest="detach_min_age",
        metavar="INTERVAL",
        help="Minimum partition age before detach consideration",
    )

    examples = [
        "absurdctl queue-policy myqueue",
        "absurdctl queue-policy myqueue --cleanup-ttl '30 days' --cleanup-limit 2000",
        "absurdctl queue-policy myqueue --default-partition disabled",
        "absurdctl queue-policy myqueue --partition-lookahead '42 days' --partition-lookback '2 days'",
        "absurdctl queue-policy myqueue --detach-mode empty --detach-min-age '30 days'",
    ]

    options, args = parse_args_with_help(parser, args, examples)

    if len(args) != 1:
        print("Error: Missing required argument QUEUE_NAME", file=sys.stderr)
        parser.print_help()
        sys.exit(1)

    queue_name = validate_queue_name(args[0])
    config = config_from_options(options)
    ensure_queue_exists(config, queue_name)

    policy_payload = {}
    if options.default_partition is not None:
        policy_payload["default_partition"] = options.default_partition
    if options.partition_lookahead is not None:
        policy_payload["partition_lookahead"] = options.partition_lookahead
    if options.partition_lookback is not None:
        policy_payload["partition_lookback"] = options.partition_lookback
    if options.cleanup_ttl is not None:
        policy_payload["cleanup_ttl"] = options.cleanup_ttl
    if options.cleanup_limit is not None:
        if options.cleanup_limit < 1:
            print("Error: --cleanup-limit must be at least 1", file=sys.stderr)
            sys.exit(1)
        policy_payload["cleanup_limit"] = options.cleanup_limit
    if options.detach_mode is not None:
        policy_payload["detach_mode"] = options.detach_mode
    if options.detach_min_age is not None:
        policy_payload["detach_min_age"] = options.detach_min_age

    print_verbose_configuration(
        options.verbose,
        [("Queue", queue_name)] + get_common_db_details(config),
    )

    if policy_payload:
        if options.verbose:
            print(f"Updating queue policy for '{queue_name}'...")
        run_psql(
            config,
            "SELECT absurd.set_queue_policy(:'queue_name', :'policy_json'::jsonb);",
            variables={
                "queue_name": queue_name,
                "policy_json": json.dumps(policy_payload),
            },
        )

    policy = _fetch_queue_policy(config, queue_name)
    if policy is None:
        print(f"Queue '{queue_name}' does not have a policy record", file=sys.stderr)
        sys.exit(1)

    _print_queue_policy(policy)


def cmd_cron(args):
    """Enable or disable Absurd-managed pg_cron jobs."""
    usage = "usage: %prog cron [options]"
    parser = build_command_parser(usage)
    parser.add_option(
        "--enable",
        dest="enable",
        action="store_true",
        default=False,
        help="Enable Absurd cron jobs",
    )
    parser.add_option(
        "--disable",
        dest="disable",
        action="store_true",
        default=False,
        help="Disable Absurd cron jobs",
    )
    parser.add_option(
        "-q",
        "--queue",
        dest="queue",
        metavar="NAME",
        help="Queue scope (omit for global/all queues scope)",
    )
    parser.add_option(
        "--partition-schedule",
        dest="partition_schedule",
        default="5 * * * *",
        metavar="CRON",
        help="Cron schedule for partition provisioning (default: '5 * * * *')",
    )
    parser.add_option(
        "--cleanup-schedule",
        dest="cleanup_schedule",
        default="17 * * * *",
        metavar="CRON",
        help="Cron schedule for cleanup maintenance (default: '17 * * * *')",
    )
    parser.add_option(
        "--detach-schedule",
        dest="detach_schedule",
        default="29 * * * *",
        metavar="CRON",
        help="Cron schedule for detach planning (default: '29 * * * *')",
    )

    examples = [
        "absurdctl cron --enable",
        "absurdctl cron --enable --queue myqueue",
        "absurdctl cron --enable --queue myqueue --partition-schedule '*/15 * * * *'",
        "absurdctl cron --disable",
        "absurdctl cron --disable --queue myqueue",
    ]

    options, args = parse_args_with_help(parser, args, examples)

    if args:
        print("Error: cron does not accept positional arguments", file=sys.stderr)
        parser.print_help()
        sys.exit(1)

    if options.enable == options.disable:
        print("Error: Provide exactly one of --enable or --disable", file=sys.stderr)
        parser.print_help()
        sys.exit(1)

    config = config_from_options(options)

    if options.queue is not None:
        options.queue = validate_queue_name(options.queue)

    print_verbose_configuration(
        options.verbose,
        [
            ("Action", "enable" if options.enable else "disable"),
            ("Queue", options.queue or "all"),
            ("Partition schedule", options.partition_schedule),
            ("Cleanup schedule", options.cleanup_schedule),
            ("Detach schedule", options.detach_schedule),
        ]
        + get_common_db_details(config),
    )

    if options.enable:
        if options.queue is None:
            query = """
                SELECT job_name, job_id
                FROM absurd.enable_cron(
                  null::text,
                  :'partition_schedule',
                  :'cleanup_schedule',
                  :'detach_schedule'
                )
                ORDER BY job_name
            """
            variables = {
                "partition_schedule": options.partition_schedule,
                "cleanup_schedule": options.cleanup_schedule,
                "detach_schedule": options.detach_schedule,
            }
        else:
            query = """
                SELECT job_name, job_id
                FROM absurd.enable_cron(
                  :'queue_name',
                  :'partition_schedule',
                  :'cleanup_schedule',
                  :'detach_schedule'
                )
                ORDER BY job_name
            """
            variables = {
                "queue_name": options.queue,
                "partition_schedule": options.partition_schedule,
                "cleanup_schedule": options.cleanup_schedule,
                "detach_schedule": options.detach_schedule,
            }

        rows = run_psql_csv(config, query, variables=variables)
        if rows:
            print("Enabled cron jobs:")
            for row in rows:
                if len(row) >= 2:
                    print(f"  {row[0]} (job_id={row[1]})")
        else:
            print("No cron jobs were created")
    else:
        if options.queue is None:
            query = """
                SELECT job_name, job_id
                FROM absurd.disable_cron(null::text)
                ORDER BY job_name
            """
            variables = None
        else:
            query = """
                SELECT job_name, job_id
                FROM absurd.disable_cron(:'queue_name')
                ORDER BY job_name
            """
            variables = {"queue_name": options.queue}

        rows = run_psql_csv(config, query, variables=variables)
        if rows:
            print("Disabled cron jobs:")
            for row in rows:
                if len(row) >= 2:
                    print(f"  {row[0]} (job_id={row[1]})")
        else:
            print("No matching cron jobs found")


def cmd_list_detach_candidates(args):
    """List partition detach candidates."""
    usage = "usage: %prog list-detach-candidates [options]"
    parser = build_command_parser(usage)
    parser.add_option(
        "-q",
        "--queue",
        dest="queue",
        metavar="NAME",
        help="Queue scope (optional)",
    )
    parser.add_option(
        "--show-sql",
        dest="show_sql",
        action="store_true",
        default=False,
        help="Include raw detach/drop SQL in output",
    )

    examples = [
        "absurdctl list-detach-candidates",
        "absurdctl list-detach-candidates --queue myqueue",
        "absurdctl list-detach-candidates --show-sql",
    ]

    options, args = parse_args_with_help(parser, args, examples)

    if args:
        print("Error: list-detach-candidates does not accept positional arguments", file=sys.stderr)
        parser.print_help()
        sys.exit(1)

    config = config_from_options(options)
    variables = None
    if options.queue is not None:
        options.queue = validate_queue_name(options.queue)
        variables = {"queue_name": options.queue}
        query = """
            SELECT
              queue_name,
              parent_table,
              partition_table
            FROM absurd.list_detach_candidates(:'queue_name')
            ORDER BY queue_name, parent_table, partition_table
        """
    else:
        query = """
            SELECT
              queue_name,
              parent_table,
              partition_table
            FROM absurd.list_detach_candidates(null::text)
            ORDER BY queue_name, parent_table, partition_table
        """

    rows = run_psql_csv(config, query, variables=variables)
    if not rows:
        print("No detach candidates found")
        return

    if not options.show_sql:
        print("partition\tqueue\tparent")
        for row in rows:
            if len(row) >= 3:
                print(f"{row[2]}\t{row[0]}\t{row[1]}")
        return

    for row in rows:
        if len(row) < 3:
            continue
        detach_sql = build_detach_partition_sql(config, row[1], row[2])
        drop_sql = build_drop_partition_sql(row[2])
        print(f"  queue: {row[0]}")
        print(f"  parent: {row[1]}")
        print(f"  partition: {row[2]}")
        print(f"  detach_sql: {detach_sql}")
        print(f"  drop_sql: {drop_sql}")
        print()


def cmd_detach_candidate(args):
    """Detach a specific candidate partition manually."""
    usage = "usage: %prog detach-candidate [options] PARTITION_TABLE"
    parser = build_command_parser(usage)
    parser.add_option(
        "-q",
        "--queue",
        dest="queue",
        metavar="NAME",
        help="Queue scope (optional)",
    )
    parser.add_option(
        "--drop",
        dest="drop",
        action="store_true",
        default=False,
        help="Attempt immediate drop after detach via absurd.drop_detached_partition",
    )

    examples = [
        "absurdctl detach-candidate t_myqueue_2414",
        "absurdctl detach-candidate --queue myqueue t_myqueue_2414 --drop",
    ]

    options, args = parse_args_with_help(parser, args, examples)

    if len(args) != 1:
        print("Error: Missing required argument PARTITION_TABLE", file=sys.stderr)
        parser.print_help()
        sys.exit(1)

    partition_table = normalize_partition_table_name(args[0])

    config = config_from_options(options)
    variables = {"partition_table": partition_table}
    if options.queue is not None:
        options.queue = validate_queue_name(options.queue)
        variables["queue_name"] = options.queue
        query = """
            SELECT
              queue_name,
              parent_table,
              partition_table
            FROM absurd.list_detach_candidates(:'queue_name')
            WHERE partition_table = :'partition_table'
            ORDER BY queue_name, parent_table, partition_table
        """
    else:
        query = """
            SELECT
              queue_name,
              parent_table,
              partition_table
            FROM absurd.list_detach_candidates(null::text)
            WHERE partition_table = :'partition_table'
            ORDER BY queue_name, parent_table, partition_table
        """

    rows = run_psql_csv(config, query, variables=variables)
    if not rows:
        print(f"No detach candidate found for partition {partition_table}", file=sys.stderr)
        sys.exit(1)

    if len(rows) > 1:
        print(
            "Error: Partition table is ambiguous across queues; rerun with --queue",
            file=sys.stderr,
        )
        for row in rows:
            if len(row) >= 3:
                print(f"  {row[0]} {row[1]} {row[2]}", file=sys.stderr)
        sys.exit(1)

    row = rows[0]
    queue_name, parent_table, partition_table = row
    detach_sql = build_detach_partition_sql(config, parent_table, partition_table)

    print(f"Detaching partition {partition_table}:")
    print(f"  queue: {queue_name}")
    print(f"  parent: {parent_table}")
    print(f"  partition: {partition_table}")
    run_psql(config, detach_sql + ";")
    print("Detach succeeded")

    if options.drop:
        dropped = run_psql(
            config,
            "SELECT absurd.drop_detached_partition(:'partition_table');",
            tuples_only=True,
            no_align=True,
            variables={"partition_table": partition_table},
        )
        if dropped.strip() == "t":
            print("Drop succeeded")
        else:
            print("Partition detached but not dropped yet")


def cmd_init(args):
    """Initialize the Absurd schema by applying absurd.sql."""
    usage = "usage: %prog init [options]"
    parser = build_command_parser(usage)
    parser.add_option(
        "--ref",
        dest="ref",
        metavar="REF",
        help=f"Git ref/tag for absurd.sql (default: {ABSURD_SCHEMA_TARGET_VERSION})",
    )

    options, args = parse_args_with_help(
        parser,
        args,
        [
            "absurdctl init",
            "absurdctl init --ref 0.1.1",
        ],
    )

    if args:
        print("Error: init does not accept positional arguments", file=sys.stderr)
        parser.print_help()
        sys.exit(1)

    config = config_from_options(options)
    target_ref = options.ref or ABSURD_SCHEMA_TARGET_VERSION

    print_verbose_configuration(
        options.verbose,
        [("Schema ref", target_ref)] + get_common_db_details(config),
    )

    script_dir = os.path.dirname(os.path.abspath(__file__))
    sql_content, sql_source = load_absurd_sql(script_dir, target_ref, options.verbose)
    result = run_psql(
        config,
        input_data=sql_content,
        error_message="Error applying SQL file",
    )

    if options.verbose and result.stdout:
        print(result.stdout)
    print(f"Absurd schema initialized successfully ({target_ref} via {sql_source})")


def cmd_schema_version(args):
    """Show the currently recorded Absurd schema version."""
    usage = "usage: %prog schema-version [options]"
    parser = build_command_parser(usage)

    options, args = parse_args_with_help(
        parser,
        args,
        ["absurdctl schema-version"],
    )

    if args:
        print("Error: schema-version does not accept positional arguments", file=sys.stderr)
        parser.print_help()
        sys.exit(1)

    config = config_from_options(options)
    version = get_recorded_schema_version(config)
    if version is None:
        print("unknown")
        return
    print(version)


def cmd_migrate(args):
    """Apply Absurd schema migrations from the recorded version to a target."""
    usage = "usage: %prog migrate [options]"
    parser = build_command_parser(usage)
    parser.add_option(
        "--from",
        dest="from_version",
        metavar="VERSION",
        help="Source schema version (overrides value recorded in DB)",
    )
    parser.add_option(
        "--to",
        dest="to_version",
        metavar="VERSION",
        help=f"Target schema version (default: {ABSURD_SCHEMA_TARGET_VERSION})",
    )
    parser.add_option(
        "--dry-run",
        dest="dry_run",
        action="store_true",
        default=False,
        help="Show the migration plan without applying SQL",
    )
    parser.add_option(
        "--dump-sql",
        dest="dump_sql",
        action="store_true",
        default=False,
        help="Print combined migration SQL for the selected range and exit",
    )

    options, args = parse_args_with_help(
        parser,
        args,
        [
            "absurdctl migrate",
            "absurdctl migrate --to 0.1.1",
            "absurdctl migrate --from 0.0.8 --to main",
            "absurdctl migrate --from 0.1.0 --to main --dump-sql > absurd-migrations.sql",
        ],
    )

    if args:
        print("Error: migrate does not accept positional arguments", file=sys.stderr)
        parser.print_help()
        sys.exit(1)

    script_dir = os.path.dirname(os.path.abspath(__file__))
    target_version = options.to_version or ABSURD_SCHEMA_TARGET_VERSION

    if options.dump_sql and options.from_version is None:
        print(
            "Error: --dump-sql requires --from VERSION so SQL can be generated from an explicit range.",
            file=sys.stderr,
        )
        sys.exit(1)

    config = None
    from_version_supplied = options.from_version is not None

    if options.dump_sql:
        current_version = options.from_version
    else:
        config = config_from_options(options)
        current_version = options.from_version or get_recorded_schema_version(config)

    if current_version is None:
        print(
            "Error: Could not determine current Absurd schema version. "
            "Pass --from VERSION, or install a schema that defines absurd.get_schema_version().",
            file=sys.stderr,
        )
        sys.exit(1)

    migrations = discover_migrations(script_dir, target_version, include_remote=False)
    path = resolve_migration_path(current_version, target_version, migrations)

    needs_remote_retry = path is None or (
        compare_versions(current_version, target_version) < 0 and not path
    )

    if needs_remote_retry:
        # Local/bundled migrations might not be complete (for example when
        # running a standalone absurdctl without migration files). Fall back to
        # GitHub to discover additional migrations.
        migrations = discover_migrations(
            script_dir,
            target_version,
            include_remote=True,
        )
        path = resolve_migration_path(current_version, target_version, migrations)

    if path is None:
        print(
            f"Error: Downgrade migrations are not supported ({current_version} -> {target_version}).",
            file=sys.stderr,
        )
        sys.exit(1)

    if not path and compare_versions(current_version, target_version) == 0:
        if not options.dump_sql:
            print(f"Absurd schema is already at {target_version}")
        return

    if not path and compare_versions(current_version, target_version) < 0:
        if not options.dump_sql:
            print(
                f"No migration files found between {current_version} and {target_version}; "
                "treating intermediate releases as null migrations."
            )
            print(f"Absurd schema migrated successfully to {target_version}")
        return

    if (options.verbose or options.dry_run) and not options.dump_sql:
        print(f"Migration plan ({current_version} -> {target_version}):")
        for migration in path:
            print(f"  - {migration['name']} [{migration['source']}]")

    if options.dry_run and not options.dump_sql:
        return

    resolved_steps = []
    for migration in path:
        sql_content = get_migration_sql(migration)
        resolved_steps.append(
            {
                "migration": migration,
                "sql": sql_content,
                "declares_schema_version": migration_declares_schema_version(
                    sql_content
                ),
            }
        )

    legacy_steps = [
        step["migration"]["name"]
        for step in resolved_steps
        if not step["declares_schema_version"]
    ]
    if legacy_steps and not from_version_supplied:
        print(
            "Error: Migration plan contains legacy migrations without "
            "absurd.get_schema_version() metadata.",
            file=sys.stderr,
        )
        print(
            "Please rerun with --from VERSION to explicitly confirm the "
            "starting schema version.",
            file=sys.stderr,
        )
        print("Legacy migrations in plan:", file=sys.stderr)
        for migration_name in legacy_steps:
            print(f"  - {migration_name}", file=sys.stderr)
        sys.exit(1)

    if options.dump_sql:
        print(
            render_combined_migration_sql(
                current_version,
                target_version,
                resolved_steps,
            ),
            end="",
        )
        return

    for step in resolved_steps:
        migration = step["migration"]
        sql_content = step["sql"]
        if options.verbose:
            print(f"Applying migration {migration['name']}...")
        run_psql(
            config,
            input_data=sql_content,
            error_message=f"Error applying migration {migration['name']}",
        )

        parsed = parse_migration_filename(migration["name"])
        if parsed is not None and step["declares_schema_version"]:
            _, migrated_to = parsed
            actual_version = get_recorded_schema_version(config)
            if actual_version != migrated_to:
                print(
                    f"Error: Migration {migration['name']} did not set schema version to {migrated_to} "
                    f"(got {actual_version or 'unknown'}).",
                    file=sys.stderr,
                )
                sys.exit(1)
        elif parsed is not None and options.verbose:
            print(
                f"Migration {migration['name']} does not define absurd.get_schema_version(); "
                "skipping version verification for this step."
            )

    print(f"Absurd schema migrated successfully to {target_version}")


def cmd_spawn_task(args):
    """Spawn a new task."""
    usage = "usage: %prog spawn-task [options] TASK_NAME"
    parser = build_command_parser(usage)

    parser.add_option(
        "-q",
        "--queue",
        dest="queue",
        metavar="NAME",
        help="Queue name (default: default)",
    )

    parser.add_option(
        "-P",
        "--param",
        dest="params",
        action="append",
        default=[],
        metavar="K=V",
        help="Task parameter. Use -P key=value for strings or -P key:=json for JSON values. "
        "Supports dotted paths for nested objects (e.g., -P user.name=Alice)",
    )
    parser.add_option(
        "--params",
        dest="base_params",
        metavar="JSON",
        help="Base JSON object for parameters (will be merged with -P flags)",
    )
    parser.add_option(
        "-H",
        "--header",
        dest="headers",
        action="append",
        default=[],
        metavar="K=V",
        help="Task header as key=value pair",
    )

    parser.add_option(
        "--max-attempts",
        dest="max_attempts",
        type="int",
        metavar="N",
        help="Maximum number of retry attempts",
    )
    parser.add_option(
        "--retry-kind",
        dest="retry_kind",
        metavar="KIND",
        help="Retry strategy: fixed, exponential, or none",
    )
    parser.add_option(
        "--retry-base",
        dest="retry_base",
        type="int",
        metavar="SECS",
        help="Base retry delay in seconds",
    )
    parser.add_option(
        "--retry-factor",
        dest="retry_factor",
        type="float",
        metavar="FACTOR",
        help="Exponential backoff multiplier (for exponential retry)",
    )
    parser.add_option(
        "--retry-max",
        dest="retry_max",
        type="int",
        metavar="SECS",
        help="Maximum retry delay cap in seconds",
    )

    parser.add_option(
        "--max-duration",
        dest="max_duration",
        type="int",
        metavar="SECS",
        help="Maximum task execution duration in seconds",
    )
    parser.add_option(
        "--max-delay",
        dest="max_delay",
        type="int",
        metavar="SECS",
        help="Maximum delay before task must start in seconds",
    )

    examples = [
        "absurdctl spawn-task my-task -P foo=bar -P count:=42",
        "absurdctl spawn-task my-task -P user.name=Alice -P user.age:=30",
        'absurdctl spawn-task my-task --params \'{"foo":"bar"}\' -P extra=value',
        "absurdctl spawn-task my-task -P email=test@example.com -q myqueue --max-attempts 5",
    ]

    options, args = parse_args_with_help(parser, args, examples)

    if len(args) != 1:
        print("Error: Missing required argument TASK_NAME", file=sys.stderr)
        parser.print_help()
        sys.exit(1)

    task_name = args[0]
    config = config_from_options(options)

    params_json = build_json_params(options.params, options.base_params)

    spawn_options = {}
    if options.headers:
        headers_dict = {}
        for header in options.headers:
            if "=" not in header:
                print(
                    f"Invalid header format: '{header}' (expected key=value)",
                    file=sys.stderr,
                )
                sys.exit(1)
            key, value = header.split("=", 1)
            headers_dict[key] = value
        spawn_options["headers"] = headers_dict

    if options.max_attempts is not None:
        spawn_options["max_attempts"] = options.max_attempts

    if (
        options.retry_kind
        or options.retry_base is not None
        or options.retry_factor is not None
        or options.retry_max is not None
    ):
        retry = {"kind": options.retry_kind or "exponential"}
        if options.retry_base is not None:
            retry["base_seconds"] = options.retry_base
        if options.retry_factor is not None:
            retry["factor"] = options.retry_factor
        if options.retry_max is not None:
            retry["max_seconds"] = options.retry_max
        spawn_options["retry_strategy"] = retry

    if options.max_duration is not None or options.max_delay is not None:
        cancellation = {}
        if options.max_duration is not None:
            cancellation["max_duration"] = options.max_duration
        if options.max_delay is not None:
            cancellation["max_delay"] = options.max_delay
        spawn_options["cancellation"] = cancellation

    options_payload = json.dumps(spawn_options)

    queue = validate_queue_name(options.queue or "default")

    ensure_queue_exists(config, queue)

    details = [
        ("Task", task_name),
        ("Queue", queue),
        ("Params", params_json),
    ]
    if "headers" in spawn_options:
        details.append(("Headers", json.dumps(spawn_options["headers"])))
    if "max_attempts" in spawn_options:
        details.append(("Max attempts", spawn_options["max_attempts"]))
    if "retry_strategy" in spawn_options:
        details.append(("Retry strategy", json.dumps(spawn_options["retry_strategy"])))
    if "cancellation" in spawn_options:
        details.append(("Cancellation", json.dumps(spawn_options["cancellation"])))
    details.extend(get_common_db_details(config))
    print_verbose_configuration(options.verbose, details)

    variables = {
        "queue": queue,
        "task_name": task_name,
        "params_json": params_json,
        "options_json": options_payload,
    }

    query = """
        SELECT task_id, run_id, attempt
        FROM absurd.spawn_task(
            :'queue',
            :'task_name',
            :'params_json'::jsonb,
            :'options_json'::jsonb
        );
    """

    rows = run_psql_csv(config, query, variables=variables)

    if rows:
        task_id, run_id, attempt = rows[0]
        print(f"Task spawned successfully:")
        print(f"  Task ID: {task_id}")
        print(f"  Run ID: {run_id}")
        print(f"  Attempt: {attempt}")
    else:
        print("Task spawned, but no result returned", file=sys.stderr)


def format_error_like(data):
    """Try to extract and format error-like objects with name, message, and stack."""
    if not isinstance(data, dict):
        return None

    if "name" in data and "message" in data and "stack" in data:
        return {
            "name": data["name"],
            "message": data["message"],
            "stack": data["stack"],
        }
    return None


def format_json_value(value, indent=0):
    """Format a JSON value with proper indentation."""
    if value is None:
        return "null"

    # Check if it's an error-like object
    error_info = format_error_like(value)
    if error_info:
        lines = []
        lines.append(f"{error_info['name']}")
        lines.append(f"\nMessage:")
        lines.append(f"  {error_info['message']}")
        if error_info["stack"]:
            lines.append(f"\nStack trace:")
            for line in error_info["stack"].split("\n"):
                lines.append(f"  {line}")
        return "\n".join(lines)

    # Otherwise, render as JSON
    return json.dumps(value, indent=2)


def format_timestamp(ts_str):
    """Format a timestamp string in a human-readable way."""
    if not ts_str:
        return "—"
    try:
        from datetime import datetime

        dt = datetime.fromisoformat(ts_str.replace("Z", "+00:00"))
        return dt.strftime("%b %d, %Y %I:%M:%S %p")
    except:
        return ts_str


def format_age(timestamp_str):
    """Format a timestamp as a relative age (e.g., '2h ago', '3d ago')."""
    if not timestamp_str:
        return "—"
    try:
        from datetime import datetime, timezone

        dt = datetime.fromisoformat(timestamp_str.replace("Z", "+00:00"))
        now = datetime.now(timezone.utc)
        delta = now - dt

        seconds = delta.total_seconds()
        if seconds < 60:
            return f"{int(seconds)}s ago"
        elif seconds < 3600:
            return f"{int(seconds / 60)}m ago"
        elif seconds < 86400:
            return f"{int(seconds / 3600)}h ago"
        elif seconds < 604800:
            return f"{int(seconds / 86400)}d ago"
        else:
            return f"{int(seconds / 604800)}w ago"
    except:
        return timestamp_str


def cmd_list_tasks(args):
    """List tasks across queues with filtering options."""
    usage = "usage: %prog list-tasks [options]"
    parser = build_command_parser(usage)
    parser.add_option(
        "-q",
        "--queue",
        dest="queue",
        metavar="NAME",
        help="Filter by queue name",
    )
    parser.add_option(
        "-n",
        "--task-name",
        dest="task_name",
        metavar="NAME",
        help="Filter by task name (exact match)",
    )
    parser.add_option(
        "-s",
        "--status",
        dest="status",
        metavar="STATUS",
        help="Filter by status (pending, running, sleeping, completed, failed, cancelled)",
    )
    parser.add_option(
        "-l",
        "--limit",
        dest="limit",
        type="int",
        default=50,
        metavar="N",
        help="Maximum number of tasks to display (default: 50)",
    )

    examples = [
        "absurdctl list-tasks",
        "absurdctl list-tasks --queue=default",
        "absurdctl list-tasks --status=failed",
        "absurdctl list-tasks --task-name=my-task --queue=default",
        "absurdctl list-tasks --limit=100",
    ]

    options, args = parse_args_with_help(parser, args, examples)

    config = config_from_options(options)

    # Validate status if provided
    valid_statuses = [
        "pending",
        "running",
        "sleeping",
        "completed",
        "failed",
        "cancelled",
    ]
    if options.status and options.status not in valid_statuses:
        print(
            f"Error: Invalid status '{options.status}'. Must be one of: {', '.join(valid_statuses)}",
            file=sys.stderr,
        )
        sys.exit(1)

    # Get list of queues
    queues_result = run_psql(
        config,
        "SELECT queue_name FROM absurd.queues;",
        tuples_only=True,
        no_align=True,
    )

    if not queues_result:
        print("No queues found in the database", file=sys.stderr)
        sys.exit(0)

    queues = [q for q in queues_result.splitlines() if q != ""]

    # Filter to specific queue if requested
    if options.queue:
        validate_queue_name(options.queue)
        if options.queue not in queues:
            print(f"Queue '{options.queue}' does not exist", file=sys.stderr)
            sys.exit(1)
        queues = [options.queue]

    # Build query for each queue and collect results
    all_tasks = []

    for queue in queues:
        # Build WHERE clause based on filters
        where_clauses = []
        variables = {}
        if options.task_name:
            where_clauses.append("t.task_name = :'task_name'")
            variables["task_name"] = options.task_name
        if options.status:
            where_clauses.append("t.state = :'status'")
            variables["status"] = options.status

        variables["queue_name"] = queue

        where_clause = ""
        if where_clauses:
            where_clause = "WHERE " + " AND ".join(where_clauses)

        t_rel = queue_relation("t", queue)
        r_rel = queue_relation("r", queue)

        query = f"""
            SELECT
                t.task_id,
                r.run_id,
                t.task_name,
                :'queue_name' as queue_name,
                t.state,
                r.attempt,
                t.max_attempts,
                COALESCE(r.completed_at, r.failed_at, r.started_at, r.created_at) AS updated_at
            FROM {t_rel} t
            LEFT JOIN {r_rel} r ON r.run_id = t.last_attempt_run
            {where_clause}
            ORDER BY updated_at DESC
            LIMIT {options.limit}
        """

        rows = run_psql_csv(config, query, variables=variables if variables else None)

        for parts in rows:
            if len(parts) >= 8:
                all_tasks.append(
                    {
                        "task_id": parts[0],
                        "run_id": parts[1] if parts[1] else "—",
                        "task_name": parts[2],
                        "queue": parts[3],
                        "status": parts[4],
                        "attempt": parts[5],
                        "max_attempts": parts[6] if parts[6] else "∞",
                        "updated_at": parts[7],
                    }
                )

    # Sort all tasks by updated_at descending
    all_tasks.sort(key=lambda x: x["updated_at"], reverse=True)

    # Apply global limit
    all_tasks = all_tasks[: options.limit]

    if not all_tasks:
        print("No tasks found")
        return

    # Print tasks with full identifiers on two lines for easier copying.
    for task in all_tasks:
        attempts = f"{task['attempt']}/{task['max_attempts']}"
        age = format_age(task["updated_at"])
        print(f"task_id={task['task_id']}")
        print(f"  run_id={task['run_id']}")
        print(f"  attempts={attempts}")
        print(f"  age={age}")
        print(f"  queue={task['queue']}")
        print(f"  task_name={task['task_name']}")
        print(f"  status={task['status']}")
        print()

    print()
    print(f"Total: {len(all_tasks)} task(s)")


def cmd_cancel_task(args):
    """Cancel a running or pending task."""
    usage = "usage: %prog cancel-task [options] TASK_ID"
    parser = build_command_parser(usage)
    parser.add_option(
        "-q",
        "--queue",
        dest="queue",
        metavar="NAME",
        help="Queue name (default: default)",
    )

    examples = [
        "absurdctl cancel-task 019a32d3-8425-7ae2-a5af-2f17a6707666",
        "absurdctl cancel-task -q myqueue 019a32d3-8425-7ae2-a5af-2f17a6707666",
    ]

    options, args = parse_args_with_help(parser, args, examples)

    if len(args) != 1:
        print("Error: Missing required argument TASK_ID", file=sys.stderr)
        parser.print_help()
        sys.exit(1)

    task_id = args[0]
    queue = validate_queue_name(options.queue or "default")
    config = config_from_options(options)

    try:
        run_psql(
            config,
            "SELECT absurd.cancel_task(:'queue', :'task_id');",
            variables={"queue": queue, "task_id": task_id},
        )
        print(f"Task {task_id} cancelled in queue '{queue}'")
    except subprocess.CalledProcessError as e:
        print(f"Error cancelling task: {e}", file=sys.stderr)
        sys.exit(1)


def cmd_retry_task(args):
    """Retry a failed task by extending attempts or spawning a new task."""
    usage = "usage: %prog retry-task [options] TASK_ID"
    parser = build_command_parser(usage)
    parser.add_option(
        "-q",
        "--queue",
        dest="queue",
        metavar="NAME",
        help="Queue name (default: default)",
    )
    parser.add_option(
        "--max-attempts",
        dest="max_attempts",
        type="int",
        metavar="N",
        help="Set max attempts for retry (in-place or when using --spawn-new)",
    )
    parser.add_option(
        "--spawn-new",
        dest="spawn_new",
        action="store_true",
        default=False,
        help="Spawn a brand new task with the original inputs",
    )

    examples = [
        "absurdctl retry-task 019a32d3-8425-7ae2-a5af-2f17a6707666",
        "absurdctl retry-task 019a32d3-8425-7ae2-a5af-2f17a6707666 --max-attempts 5",
        "absurdctl retry-task -q myqueue 019a32d3-8425-7ae2-a5af-2f17a6707666 --spawn-new",
        "absurdctl retry-task 019a32d3-8425-7ae2-a5af-2f17a6707666 --spawn-new --max-attempts 8",
    ]

    options, args = parse_args_with_help(parser, args, examples)

    if len(args) != 1:
        print("Error: Missing required argument TASK_ID", file=sys.stderr)
        parser.print_help()
        sys.exit(1)

    task_id = args[0]
    queue = validate_queue_name(options.queue or "default")
    config = config_from_options(options)

    retry_options = {}
    if options.spawn_new:
        retry_options["spawn_new"] = True
    if options.max_attempts is not None:
        retry_options["max_attempts"] = options.max_attempts

    rows = run_psql_csv(
        config,
        """
        SELECT task_id, run_id, attempt, created
        FROM absurd.retry_task(:'queue', :'task_id', :'options_json'::jsonb);
        """,
        variables={
            "queue": queue,
            "task_id": task_id,
            "options_json": json.dumps(retry_options),
        },
    )

    if not rows:
        print("Retry requested, but no result returned", file=sys.stderr)
        sys.exit(1)

    out_task_id, out_run_id, out_attempt, out_created = rows[0]
    created = str(out_created).lower() in ("t", "true", "1")

    if created:
        print("Task respawned successfully:")
    else:
        print("Task retried successfully:")
    print(f"  Task ID: {out_task_id}")
    print(f"  Run ID: {out_run_id}")
    print(f"  Attempt: {out_attempt}")


def cmd_dump_task(args):
    """Dump task or run details including checkpoints."""
    usage = "usage: %prog dump-task [options]"
    parser = build_command_parser(usage)
    parser.add_option(
        "--task-id",
        dest="task_id",
        metavar="UUID",
        help="Task ID to dump",
    )
    parser.add_option(
        "--run-id",
        dest="run_id",
        metavar="UUID",
        help="Run ID to dump",
    )

    examples = [
        "absurdctl dump-task --task-id=019a32d3-8425-7ae2-a5af-2f17a6707666",
        "absurdctl dump-task --run-id=019a32d3-8425-7ae2-a5af-2f17a6707666",
    ]

    options, args = parse_args_with_help(parser, args, examples)

    if not options.task_id and not options.run_id:
        print("Error: Either --task-id or --run-id must be specified", file=sys.stderr)
        parser.print_help()
        sys.exit(1)

    if options.task_id and options.run_id:
        print("Error: Cannot specify both --task-id and --run-id", file=sys.stderr)
        parser.print_help()
        sys.exit(1)

    config = config_from_options(options)

    # Get list of queues
    queues_result = run_psql(
        config,
        "SELECT queue_name FROM absurd.queues;",
        tuples_only=True,
        no_align=True,
    )

    if not queues_result:
        print("No queues found in the database", file=sys.stderr)
        sys.exit(1)

    queues = [q for q in queues_result.splitlines() if q != ""]

    # Find which queue contains the task/run
    target_queue = None
    task_id = None

    if options.run_id:
        # Search for run_id across all queues
        for queue in queues:
            r_rel = queue_relation("r", queue)
            result = run_psql(
                config,
                f"SELECT task_id FROM {r_rel} WHERE run_id = :'run_id';",
                tuples_only=True,
                no_align=True,
                variables={"run_id": options.run_id},
            )
            if result:
                target_queue = queue
                task_id = result.strip()
                break

        if not target_queue:
            print(f"Run ID '{options.run_id}' not found in any queue", file=sys.stderr)
            sys.exit(1)
    else:
        # Search for task_id across all queues
        task_id = options.task_id
        for queue in queues:
            t_rel = queue_relation("t", queue)
            result = run_psql(
                config,
                f"SELECT COUNT(*) FROM {t_rel} WHERE task_id = :'task_id';",
                tuples_only=True,
                no_align=True,
                variables={"task_id": task_id},
            )
            if result and int(result.strip()) > 0:
                target_queue = queue
                break

        if not target_queue:
            print(f"Task ID '{task_id}' not found in any queue", file=sys.stderr)
            sys.exit(1)

    t_target_rel = queue_relation("t", target_queue)
    r_target_rel = queue_relation("r", target_queue)

    # Fetch task and runs
    if options.run_id:
        # Fetch specific run
        runs_query = f"""
            SELECT
                t.task_id,
                r.run_id,
                t.task_name,
                t.state,
                r.state as run_state,
                r.attempt,
                t.max_attempts,
                t.params,
                t.retry_strategy,
                t.headers,
                COALESCE(r.failure_reason, r.result) AS final_state,
                r.created_at,
                r.started_at,
                r.completed_at,
                r.failed_at,
                t.cancelled_at,
                COALESCE(r.completed_at, r.failed_at, r.started_at, r.created_at) AS updated_at,
                r.claimed_by
            FROM {t_target_rel} t
            JOIN {r_target_rel} r ON r.task_id = t.task_id
            WHERE r.run_id = :'run_id'
        """
        runs_rows = run_psql_csv(config, runs_query, variables={"run_id": options.run_id})
    else:
        # Fetch all runs for task
        runs_query = f"""
            SELECT
                t.task_id,
                r.run_id,
                t.task_name,
                t.state,
                r.state as run_state,
                r.attempt,
                t.max_attempts,
                t.params,
                t.retry_strategy,
                t.headers,
                COALESCE(r.failure_reason, r.result) AS final_state,
                r.created_at,
                r.started_at,
                r.completed_at,
                r.failed_at,
                t.cancelled_at,
                COALESCE(r.completed_at, r.failed_at, r.started_at, r.created_at) AS updated_at,
                r.claimed_by
            FROM {t_target_rel} t
            JOIN {r_target_rel} r ON r.task_id = t.task_id
            WHERE t.task_id = :'task_id'
            ORDER BY r.attempt
        """
        runs_rows = run_psql_csv(config, runs_query, variables={"task_id": task_id})

    if not runs_rows:
        print("No data found", file=sys.stderr)
        sys.exit(1)

    # Parse runs
    runs = []
    for parts in runs_rows:
        if len(parts) >= 18:
            runs.append(
                {
                    "task_id": parts[0],
                    "run_id": parts[1],
                    "task_name": parts[2],
                    "task_state": parts[3],
                    "run_state": parts[4],
                    "attempt": parts[5],
                    "max_attempts": parts[6] if parts[6] else None,
                    "params": json.loads(parts[7]) if parts[7] else None,
                    "retry_strategy": json.loads(parts[8]) if parts[8] else None,
                    "headers": json.loads(parts[9]) if parts[9] else None,
                    "final_state": json.loads(parts[10]) if parts[10] else None,
                    "created_at": parts[11],
                    "started_at": parts[12],
                    "completed_at": parts[13],
                    "failed_at": parts[14],
                    "cancelled_at": parts[15],
                    "updated_at": parts[16],
                    "worker_id": parts[17] if parts[17] else None,
                }
            )

    c_target_rel = queue_relation("c", target_queue)
    w_target_rel = queue_relation("w", target_queue)
    e_target_rel = queue_relation("e", target_queue)

    # For each run, fetch checkpoints and waits
    for run in runs:
        # Fetch checkpoints
        checkpoints_query = f"""
            SELECT
                checkpoint_name,
                state,
                status,
                owner_run_id,
                updated_at
            FROM {c_target_rel}
            WHERE task_id = :'task_id' AND owner_run_id = :'run_id'
            ORDER BY updated_at DESC
        """

        checkpoint_rows = run_psql_csv(
            config, checkpoints_query, variables={"task_id": run["task_id"], "run_id": run["run_id"]}
        )

        checkpoints = []
        for parts in checkpoint_rows:
            if len(parts) >= 5:
                checkpoints.append(
                    {
                        "name": parts[0],
                        "state": json.loads(parts[1]) if parts[1] else None,
                        "status": parts[2],
                        "owner_run_id": parts[3],
                        "updated_at": parts[4],
                    }
                )

        run["checkpoints"] = checkpoints

        # Fetch wait states
        waits_query = f"""
            SELECT
                CASE
                    WHEN r.wake_event IS NOT NULL THEN 'event'
                    WHEN r.available_at IS NOT NULL THEN 'sleep'
                    ELSE NULL
                END AS wait_type,
                r.available_at,
                r.wake_event,
                w.step_name,
                r.event_payload,
                e.payload as event_payload_emitted,
                e.emitted_at,
                COALESCE(w.created_at, r.started_at) as updated_at
            FROM {r_target_rel} r
            LEFT JOIN {w_target_rel} w ON w.run_id = r.run_id
            LEFT JOIN {e_target_rel} e ON e.event_name = r.wake_event AND e.payload IS NOT NULL
            WHERE r.run_id = :'run_id' AND r.state = 'sleeping'
        """

        waits_rows = run_psql_csv(config, waits_query, variables={"run_id": run["run_id"]})

        waits = []
        for parts in waits_rows:
            if len(parts) >= 8 and parts[0]:
                waits.append(
                    {
                        "wait_type": parts[0],
                        "wake_at": parts[1],
                        "wake_event": parts[2] if parts[2] else None,
                        "step_name": parts[3] if parts[3] else None,
                        "payload": json.loads(parts[4]) if parts[4] else None,
                        "event_payload": json.loads(parts[5]) if parts[5] else None,
                        "emitted_at": parts[6] if parts[6] else None,
                        "updated_at": parts[7],
                    }
                )

        run["waits"] = waits

    # Now output the results
    print("=" * 80)
    print(f"TASK DUMP")
    print("=" * 80)
    print()

    for idx, run in enumerate(runs):
        if idx > 0:
            print()
            print("-" * 80)
            print()

        print(f"BASIC INFORMATION")
        print(f"  Current status:    {run['run_state']}")
        print(f"  Task Name:         {run['task_name']}")
        print(f"  Queue:             {target_queue}")
        print(f"  Task ID:           {run['task_id']}")
        print(f"  Run ID:            {run['run_id']}")
        if run["worker_id"]:
            print(f"  Worker ID:         {run['worker_id']}")
        print()

        print(f"TIMING")
        print(f"  Created:           {format_timestamp(run['created_at'])}")
        print(f"  Updated:           {format_timestamp(run['updated_at'])}")
        if run["started_at"]:
            print(f"  Started:           {format_timestamp(run['started_at'])}")
        if run["completed_at"]:
            print(f"  Completed:         {format_timestamp(run['completed_at'])}")
        if run["failed_at"]:
            print(f"  Failed:            {format_timestamp(run['failed_at'])}")
        if run["cancelled_at"]:
            print(f"  Cancelled:         {format_timestamp(run['cancelled_at'])}")
        print()

        # Wait states
        if run["waits"]:
            print(f"WAIT STATES")
            for wait in run["waits"]:
                wait_type = (
                    wait["wait_type"].capitalize() + " wait"
                    if wait["wait_type"]
                    else "Wait"
                )
                print(f"  {wait_type}")
                if wait["step_name"]:
                    print(f"    Step:            {wait['step_name']}")
                if wait["wake_at"]:
                    print(f"    Wake at:         {format_timestamp(wait['wake_at'])}")
                if wait["wake_event"]:
                    print(f"    Wake event:      {wait['wake_event']}")
                if wait["emitted_at"]:
                    print(
                        f"    Last emit:       {format_timestamp(wait['emitted_at'])}"
                    )
                print(f"    Updated:         {format_timestamp(wait['updated_at'])}")
                if wait["payload"]:
                    print(f"    Wait payload:")
                    for line in format_json_value(wait["payload"]).split("\n"):
                        print(f"      {line}")
                if wait["event_payload"]:
                    print(f"    Event payload:")
                    for line in format_json_value(wait["event_payload"]).split("\n"):
                        print(f"      {line}")
            print()

        print(f"RETRY INFORMATION")
        max_attempts_str = run["max_attempts"] if run["max_attempts"] else "unlimited"
        print(f"  Attempt:           {run['attempt']} of {max_attempts_str}")
        if run["retry_strategy"]:
            print(f"  Retry Strategy:")
            for line in format_json_value(run["retry_strategy"]).split("\n"):
                print(f"    {line}")
        print()

        # Parameters
        if run["params"]:
            print(f"PARAMETERS")
            for line in format_json_value(run["params"]).split("\n"):
                print(f"  {line}")
            print()

        # Headers
        if run["headers"]:
            print(f"HEADERS")
            for line in format_json_value(run["headers"]).split("\n"):
                print(f"  {line}")
            print()

        # Final state
        if run["final_state"]:
            state_label = "FAILURE" if run["run_state"] == "failed" else "FINAL STATE"
            print(state_label)
            for line in format_json_value(run["final_state"]).split("\n"):
                print(f"  {line}")
            print()

        # Checkpoints
        if run["checkpoints"]:
            print(f"CHECKPOINTS")
            for checkpoint in run["checkpoints"]:
                print(f"  {checkpoint['name']} ({checkpoint['status']})")
                print(
                    f"    Updated:         {format_timestamp(checkpoint['updated_at'])}"
                )
                if checkpoint["state"]:
                    print(f"    State:")
                    for line in format_json_value(checkpoint["state"]).split("\n"):
                        print(f"      {line}")
                print()


def cmd_emit_event(args):
    """Emit an event to wake waiting tasks."""
    usage = "usage: %prog emit-event [options] EVENT_NAME"
    parser = build_command_parser(usage)

    parser.add_option(
        "-q",
        "--queue",
        dest="queue",
        metavar="NAME",
        help="Queue name (default: default)",
    )

    parser.add_option(
        "-P",
        "--param",
        dest="params",
        action="append",
        default=[],
        metavar="K=V",
        help="Event payload parameter. Use -P key=value for strings or -P key:=json for JSON values. "
        "Supports dotted paths for nested objects (e.g., -P user.name=Alice)",
    )
    parser.add_option(
        "--payload",
        dest="base_payload",
        metavar="JSON",
        help="Base JSON object for payload (will be merged with -P flags)",
    )

    examples = [
        "absurdctl emit-event order.completed",
        "absurdctl emit-event order.completed -P orderId=123 -P status=shipped",
        "absurdctl emit-event user.created -P user.id:=42 -P user.name=Alice",
        'absurdctl emit-event order.completed --payload \'{"orderId":"123"}\'',
        "absurdctl emit-event -q myqueue shipment.packed -P trackingNumber=ABC123",
    ]

    options, args = parse_args_with_help(parser, args, examples)

    if len(args) != 1:
        print("Error: Missing required argument EVENT_NAME", file=sys.stderr)
        parser.print_help()
        sys.exit(1)

    event_name = args[0]
    config = config_from_options(options)

    payload_json = build_json_params(options.params, options.base_payload)

    queue = validate_queue_name(options.queue or "default")

    ensure_queue_exists(config, queue)

    details = [
        ("Event", event_name),
        ("Queue", queue),
        ("Payload", payload_json),
    ]
    details.extend(get_common_db_details(config))
    print_verbose_configuration(options.verbose, details)

    query = "SELECT absurd.emit_event(:'queue', :'event_name', :'payload_json'::jsonb);"
    run_psql(
        config,
        query,
        variables={
            "queue": queue,
            "event_name": event_name,
            "payload_json": payload_json,
        },
    )
    print(f"Event '{event_name}' emitted successfully to queue '{queue}'")


def cmd_install_skill(args):
    """Install the bundled Absurd agent skill into a skills directory."""
    usage = "usage: %prog install-skill [options] [TARGET_DIR]"
    parser = OptionParser(
        usage=usage,
        add_help_option=False,
        formatter=CleanHelpFormatter(),
    )
    parser.add_option(
        "--force",
        dest="force",
        action="store_true",
        default=False,
        help="Overwrite an existing installed skill if it differs",
    )
    parser.add_option(
        "--help",
        action="store_true",
        help="Show this help message",
    )

    examples = [
        "absurdctl install-skill",
        "absurdctl install-skill .pi/skills",
        "absurdctl install-skill ~/.pi/agent/skills --force",
    ]

    options, args = parse_args_with_help(parser, args, examples)

    if len(args) > 1:
        print("Error: install-skill accepts at most one positional TARGET_DIR", file=sys.stderr)
        parser.print_help()
        sys.exit(1)

    target_root = os.path.expanduser(args[0] if args else ".agents/skills")
    skill_name = "absurd"
    relative_skill_path = f"skills/{skill_name}/SKILL.md"
    script_dir = os.path.dirname(os.path.abspath(__file__))
    skill_content, skill_source = load_skill_file(script_dir, relative_skill_path, False)

    destination = os.path.join(target_root, skill_name, "SKILL.md")
    destination_dir = os.path.dirname(destination)
    os.makedirs(destination_dir, exist_ok=True)

    if os.path.exists(destination):
        with open(destination, "r", encoding="utf-8") as f:
            existing = f.read()
        if existing == skill_content:
            print(f"Skill '{skill_name}' is already installed at {destination}")
            return
        if not options.force:
            print(
                f"Error: {destination} already exists and differs. Re-run with --force to overwrite.",
                file=sys.stderr,
            )
            sys.exit(1)

    with open(destination, "w", encoding="utf-8") as f:
        f.write(skill_content)

    print(f"Installed skill '{skill_name}' to {destination} ({skill_source})")


def show_help():
    print(
        """Usage: absurdctl COMMAND [OPTIONS]

A command-line utility for managing Absurd queues.

Commands:
  init            Initialize the Absurd schema by applying absurd.sql
  migrate         Apply schema migrations to a target version
  schema-version  Show the recorded Absurd schema version
  cleanup         Clean up old completed, failed, or cancelled tasks and events
  create-queue    Create a new queue
  queue-policy    Get or update queue maintenance policy
  cron            Enable or disable Absurd-managed pg_cron jobs
  list-detach-candidates  List eligible partition detach candidates
  detach-candidate        Detach one candidate partition manually
  drop-queue      Drop an existing queue
  list-queues     List all existing queues
  spawn-task      Spawn a new task
  retry-task      Retry a failed task
  emit-event      Emit an event to wake waiting tasks
  list-tasks      List tasks with optional filtering
  cancel-task     Cancel a running or pending task
  dump-task       Dump task or run details including checkpoints
  install-skill   Install the bundled Absurd agent skill into a skills folder
  help            Show this help message

Run 'absurdctl COMMAND --help' for more information on a command.

Environment Variables:
  ABSURD_DATABASE_URL  PostgreSQL connection URI (checked before PGDATABASE)
  PGHOST          Database host
  PGPORT          Database port
  PGUSER          Database user
  PGDATABASE      Database name or full PostgreSQL connection URI
  PGPASSWORD      Database password (recommended over command line)

Examples:
  export PGDATABASE="postgresql://user:pass@localhost:5432/mydb"
  absurdctl init
  absurdctl migrate
  absurdctl schema-version
  absurdctl cleanup myqueue 30
  absurdctl create-queue myqueue --storage-mode partitioned
  absurdctl queue-policy myqueue --cleanup-ttl '7 days'
  absurdctl cron --enable --queue myqueue
  absurdctl list-detach-candidates --queue myqueue
  absurdctl detach-candidate --queue myqueue t_myqueue_2414 --drop
  absurdctl drop-queue myqueue --yes
  absurdctl list-queues
  absurdctl spawn-task my-task -P foo=bar -P count:=42
  absurdctl retry-task 019a32d3-8425-7ae2-a5af-2f17a6707666 --max-attempts 5
  absurdctl emit-event order.completed -P orderId=123
  absurdctl list-tasks --queue=default --status=failed
  absurdctl cancel-task 019a32d3-8425-7ae2-a5af-2f17a6707666
  absurdctl dump-task --task-id=019a32d3-8425-7ae2-a5af-2f17a6707666
  absurdctl install-skill .pi/skills
"""
    )


def main():
    if len(sys.argv) < 2 or sys.argv[1] in ("help", "--help", "-h"):
        show_help()
        sys.exit(0)

    command = sys.argv[1]
    args = sys.argv[2:]

    if command == "cleanup":
        cmd_cleanup(args)
    elif command == "migrate":
        cmd_migrate(args)
    elif command == "schema-version":
        cmd_schema_version(args)
    elif command == "create-queue":
        cmd_create_queue(args)
    elif command == "queue-policy":
        cmd_queue_policy(args)
    elif command == "cron":
        cmd_cron(args)
    elif command == "list-detach-candidates":
        cmd_list_detach_candidates(args)
    elif command == "detach-candidate":
        cmd_detach_candidate(args)
    elif command == "drop-queue":
        cmd_drop_queue(args)
    elif command == "list-queues":
        cmd_list_queues(args)
    elif command == "init":
        cmd_init(args)
    elif command == "spawn-task":
        cmd_spawn_task(args)
    elif command == "retry-task":
        cmd_retry_task(args)
    elif command == "emit-event":
        cmd_emit_event(args)
    elif command == "list-tasks":
        cmd_list_tasks(args)
    elif command == "cancel-task":
        cmd_cancel_task(args)
    elif command == "dump-task":
        cmd_dump_task(args)
    elif command == "install-skill":
        cmd_install_skill(args)
    else:
        print(f"Unknown command: {command}", file=sys.stderr)
        print("")
        show_help()
        sys.exit(1)


if __name__ == "__main__":
    main()