scripts.minimax_irc_task_picker

Reference › scripts › minimax_irc_task_picker

scripts/minimax_irc_task_picker.py is the task-queue picker for the minimax agent on the IRC channel. It reads a JSON-lines task log (one JSON object per line; _load_tasks skips blank and malformed lines) and lets the agent list pending tasks and report on them. A human or the minimax agent runs it; nothing imports it.

There are two subcommands. list (cmd_list) shows pending tasks, --all includes tasks already picked up, and --mark-picked marks every currently pending task as picked_up. report (cmd_report) appends a message to one task, given --task-id and --message. Both write back through _rewrite, which rewrites the task log. The listener side of the same IRC tooling is described by _irc_listen.

At a glance

Class diagram

No classes; see Functions.

Dependency diagram

No internal imports and no importers.

API

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/scripts/minimax_irc_task_picker.py

#!/usr/bin/env python3
"""minimax IRC task queue picker.

Reads pending tasks from `~/.local/state/minimax/tasks.jsonl` (written by
the `minimax-irc-bot.py` daemon). Each task is a message from `claude`
mentioning `minimax`; the bot auto-acks on IRC and logs here.

The opencode model runs this at session start to see what claude sent since
the last session:

    python scripts/minimax_irc_task_picker.py            # pending only
    python scripts/minimax_irc_task_picker.py --all      # everything
    python scripts/minimax_irc_task_picker.py --mark-picked

`--mark-picked` flips `picked_up=true` for all currently-pending rows
(atomic file rewrite, flock-protected). Reports get appended back via
`--report <task_id> --message "..."`.

The bot honours a push-authorisation heuristic: if claude's raw message
mentions `push`/`merge`/`force-push`, the row carries `push_authorized: true`;
otherwise the model must NOT push (per the user's directive).
"""

from __future__ import annotations

import argparse
import fcntl
import json
import os
import re
import sys
from pathlib import Path

STATE_DIR = Path(
    os.environ.get("MINIMAX_STATE_DIR", Path.home() / ".local" / "state" / "minimax")
)
TASKS_LOG = STATE_DIR / "tasks.jsonl"

PUSH_HINT_RE = re.compile(r"(?i)\b(push(\s+it)?|merge(\s+it)?|force[-_ ]?push)\b")


def _load_tasks() -> list[dict]:
    if not TASKS_LOG.exists():
        return []
    out: list[dict] = []
    with open(TASKS_LOG, encoding="utf-8") as f:
        for raw in f:
            raw = raw.strip()
            if not raw:
                continue
            try:
                out.append(json.loads(raw))
            except json.JSONDecodeError:
                continue
    return out


def _rewrite(tasks: list[dict]) -> None:
    tmp = TASKS_LOG.with_suffix(".tmp")
    with open(tmp, "w", encoding="utf-8") as f:
        fcntl.flock(f.fileno(), fcntl.LOCK_EX)
        try:
            for t in tasks:
                f.write(json.dumps(t, ensure_ascii=False) + "\n")
        finally:
            fcntl.flock(f.fileno(), fcntl.LOCK_UN)
    os.replace(tmp, TASKS_LOG)


def cmd_list(args: argparse.Namespace) -> int:
    tasks = _load_tasks()
    if not tasks:
        print(f"# No tasks in {TASKS_LOG}")
        return 0

    shown = tasks if args.all else [t for t in tasks if not t.get("picked_up")]
    pending = sum(1 for t in tasks if not t.get("picked_up"))
    total = len(tasks)

    if not shown:
        print(f"# 0 pending / {total} total — nothing to do.")
        return 0

    header = "# all" if args.all else f"# {pending} pending / {total} total"
    print(header)
    print()

    for t in shown:
        push = "PUSH-AUTH" if t.get("push_authorized") else "no-push"
        flags = f"{push} picked_up={t.get('picked_up', False)}"
        print(f"## {t['task_id']}  {t['ts']}  from {t['from']}  [{flags}]")
        print(f"Summary: {t.get('summary', '(none)')}")
        print()
        print(t["raw"])
        print()
        print("---")

    if args.mark_picked and not args.all:
        ids = {t["task_id"] for t in shown}
        for t in tasks:
            if t["task_id"] in ids:
                t["picked_up"] = True
        _rewrite(tasks)
        print(f"\nMarked {len(ids)} task(s) as picked_up.")

    return 0


def cmd_report(args: argparse.Namespace) -> int:
    tasks = _load_tasks()
    target = next((t for t in tasks if t["task_id"] == args.task_id), None)
    if target is None:
        print(f"task_id {args.task_id!r} not found in {TASKS_LOG}", file=sys.stderr)
        return 1
    target["report"] = {
        "ts": __import__("datetime")
        .datetime.now(__import__("datetime").timezone.utc)
        .isoformat(timespec="seconds"),
        "message": args.message,
    }
    target["picked_up"] = True
    _rewrite(tasks)
    print(f"reported on {args.task_id}: {args.message[:120]}")
    return 0


def main() -> int:
    p = argparse.ArgumentParser(
        description=__doc__.splitlines()[0] if __doc__ else None
    )
    sub = p.add_subparsers(dest="cmd", required=True)

    p_list = sub.add_parser("list", help="List tasks (default: pending only).")
    p_list.add_argument(
        "--all", action="store_true", help="Show all tasks including already-picked-up."
    )
    p_list.add_argument(
        "--mark-picked",
        action="store_true",
        help="Mark all currently-pending tasks as picked_up.",
    )
    p_list.set_defaults(func=cmd_list)

    p_rep = sub.add_parser("report", help="Append a report to a task.")
    p_rep.add_argument("--task-id", required=True)
    p_rep.add_argument("--message", required=True)
    p_rep.set_defaults(func=cmd_report)

    args = p.parse_args()
    return args.func(args)


if __name__ == "__main__":
    sys.exit(main())