agent-home-kit/bin/dispatch.py

#!/usr/bin/env python3
"""Диспетчер «одного окна» — минимальное ядро (agent-home-kit; исходник — bin/dispatch.py дома Дистилла).

Окно-координатор ставит задачу файлом ТЗ, диспетчер заносит её в реестр (state/dispatch/registry.jsonl),
запускает виток на узле пула (kit.json: nodes, runner), виток сдаёт через witness-log --done и пишет
итог координатору по шине; приёмка — accept.py по блоку «Приёмка:» ТЗ.

  dispatch.py add <метка> <проект> <ТЗ> --reply <метка координатора> "<суть>" [--goal <блок>]
  dispatch.py list [--open]          задачи: метка · проект · узел · сдано · вердикт
  dispatch.py launch <метка> [--host <узел>]   запустить виток (runner из kit.json); без runner — печать команды
  dispatch.py mail                   сдачи без вердикта приёмки (для ступени mail лестницы)
  dispatch.py accept <метка>         то же, что accept.py check <метка>

Формат реестра совместим с домом Дистилла: запись задачи без поля ev (at/label/project/tz/reply/what[/goal]),
события — с полем ev (launch, accept).
"""
from __future__ import annotations

import argparse
import datetime as dt
import json
import shlex
import subprocess
import sys
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent))
from kit import CFG, HOME, AGENT  # noqa: E402
import importlib.util as _iu  # noqa: E402

_s = _iu.spec_from_file_location('accept_mod', str(HOME / 'bin/accept.py'))
acc = _iu.module_from_spec(_s); _s.loader.exec_module(acc)

ST = HOME / 'state/dispatch'
REG = ST / 'registry.jsonl'
RUNS = HOME / 'var/runs'

BODY = """Ты — {agent}. Сессия запущена диспетчером на узле {host}: headless.
Метка сессии: {label}
ТЗ: {tz}
Проект: {project}
Ответ: {reply}

Порядок:
1. Лестница пробуждения, по одной ступени за вызов, до строки «ЛЕСТНИЦА ПРОЙДЕНА»:
   python3 {home}/bin/wake-ladder.py next {label} --project "{project}"
2. Прочитай ТЗ {home}/{tz} и исполняй его до результата: один предмет за ход, файлы по одному и по делу.
3. Сдача: python3 {home}/bin/witness-log.py --surface headless --session {label} --item <пункт> --done "<предмет → результат>"
   Итог 3–5 строками координатору: python3 {home}/bin/session-bus.py send --to {reply} --session {label} "<итог>"
"""


def iso() -> str:
    return dt.datetime.now(dt.timezone.utc).strftime('%Y-%m-%dT%H:%M:%SZ')


def rows() -> list[dict]:
    return acc._jsonl(REG)


def put(rec: dict) -> None:
    ST.mkdir(parents=True, exist_ok=True)
    with REG.open('a', encoding='utf-8') as f:
        f.write(json.dumps(rec, ensure_ascii=False) + '\n')


def tasks() -> dict[str, dict]:
    """Свёртка реестра по метке: задача + последний launch + последний accept + сдано ли."""
    out: dict[str, dict] = {}
    for r in rows():
        lb = r.get('label')
        if not lb:
            continue
        if 'ev' not in r:
            out[lb] = {**r, 'host': '', 'verdict': ''}
        elif lb in out and r['ev'] == 'launch':
            out[lb]['host'] = r.get('host', '')
        elif lb in out and r['ev'] == 'accept':
            out[lb]['verdict'] = r.get('verdict', '')
    done = {r.get('session') for r in acc._jsonl(acc.TURNS) if r.get('kind') == 'done'}
    for lb, t in out.items():
        t['done'] = lb in done
    return out


def cmd_add(a) -> int:
    if a.label in tasks():
        print(f'{a.label}: уже в реестре'); return 1
    ok, msg = acc.gate(a.tz)
    if not ok:
        print(msg); return 1
    rec = {'at': iso(), 'label': a.label, 'project': a.project, 'tz': a.tz, 'reply': a.reply, 'what': a.what}
    if a.goal:
        rec['goal'] = a.goal
    put(rec)
    print(f'{a.label}: в реестре; {msg}')
    return 0


def cmd_list(a) -> int:
    ts = tasks()
    n = 0
    for lb, t in ts.items():
        if a.open and t['verdict'] == 'ok':
            continue
        n += 1
        print(f"{lb:<28} {t.get('project',''):<12} узел={t['host'] or '-':<8} сдано={'да' if t['done'] else 'нет':<3} вердикт={t['verdict'] or '-'}  {t.get('what','')[:50]}")
    if not n:
        print('задач нет' if not a.open else 'открытых задач нет')
    return 0


def pick_host(want: str) -> str:
    nodes = CFG.get('nodes') or {}
    if want:
        if want not in nodes:
            raise SystemExit(f'узла {want} нет в kit.json: nodes ({", ".join(nodes)})')
        return want
    busy: dict[str, int] = {}
    for t in tasks().values():
        if t['host'] and not t['done']:
            busy[t['host']] = busy.get(t['host'], 0) + 1
    for h, spec in nodes.items():
        if busy.get(h, 0) < int(spec.get('max_runs', 1)):
            return h
    raise SystemExit('пул занят: все узлы kit.json на пределе max_runs')


def cmd_launch(a) -> int:
    ts = tasks()
    if a.label not in ts:
        print(f'{a.label}: в реестре нет'); return 1
    t = ts[a.label]
    host = pick_host(a.host)
    body = BODY.format(agent=AGENT, host=host, label=a.label, tz=t['tz'], project=t.get('project', ''),
                       reply=t.get('reply', ''), home=HOME)
    RUNS.mkdir(parents=True, exist_ok=True)
    bf = RUNS / f'{a.label}.prompt.md'
    bf.write_text(body, encoding='utf-8')
    runner = CFG.get('runner', '')
    if not runner:
        print(f'runner в kit.json не задан; промпт витка: {bf}\nпример: claude -p "$(cat {bf})"')
        return 0
    cmd = runner.format(prompt=bf, label=a.label, tz=t['tz'], home=HOME)
    ssh = (CFG['nodes'].get(host) or {}).get('ssh', '')
    argv = (['ssh', ssh, cmd] if ssh else shlex.split(cmd))
    log = (RUNS / f'{a.label}.log').open('a', encoding='utf-8')
    subprocess.Popen(argv, stdout=log, stderr=subprocess.STDOUT, cwd=str(HOME), start_new_session=True)
    put({'ev': 'launch', 'label': a.label, 'host': host, 'at': iso()})
    print(f'{a.label}: запущен на {host}; журнал {RUNS / (a.label + ".log")}')
    return 0


def cmd_mail(a) -> int:
    waiting = [lb for lb, t in tasks().items() if t['done'] and not t['verdict']]
    print(f'диспетчер: сдано без приёмки {len(waiting)}' + (' → ' + ', '.join(waiting[:5]) + ' (dispatch.py accept <метка>)' if waiting else ''))
    return 0


def main() -> int:
    p = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
    s = p.add_subparsers(dest='c', required=True)
    x = s.add_parser('add'); x.add_argument('label'); x.add_argument('project'); x.add_argument('tz')
    x.add_argument('what'); x.add_argument('--reply', required=True); x.add_argument('--goal', default='')
    x.set_defaults(fn=cmd_add)
    x = s.add_parser('list'); x.add_argument('--open', action='store_true'); x.set_defaults(fn=cmd_list)
    x = s.add_parser('launch'); x.add_argument('label'); x.add_argument('--host', default=''); x.set_defaults(fn=cmd_launch)
    s.add_parser('mail').set_defaults(fn=cmd_mail)
    x = s.add_parser('accept'); x.add_argument('label'); x.set_defaults(fn=lambda a: acc.cmd_check(a.label))
    a = p.parse_args()
    return a.fn(a)


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