agent-home-kit/bin/accept.py

#!/usr/bin/env python3
"""Приёмка сдач по блоку «Приёмка:» ТЗ — zero-LLM ядро (agent-home-kit; исходник — bin/accept.py дома Дистилла).

Пункт блока проверяется по виду: URL отдаёт код, файл есть и не пуст, коммит есть в git дома,
число в сдаче удовлетворяет порогу, письмо координатору не длиннее N строк. Пункт без
проверяемого условия уходит судье: в наборе судья — человек или ваша модель (вердикт judge).

  accept.py gate <ТЗ>        есть ли в ТЗ блок приёмки с проверяемым пунктом (гейт dispatch.py add)
  accept.py check <метка>    проверить сдачу витка <метка>: ev=accept ok|return|judge в реестр
  accept.py selftest         проба на временном доме

Сдача витка: строки witness-log --done этой метки (state/witness/turns.jsonl) + письма шины от неё.
"""
from __future__ import annotations

import json
import pathlib
import re
import subprocess
import sys
import urllib.error
import urllib.request
import datetime as dt

sys.path.insert(0, str(pathlib.Path(__file__).resolve().parent))
from kit import HOME as H  # noqa: E402

REG = H / 'state/dispatch/registry.jsonl'
TURNS = H / 'state/witness/turns.jsonl'
BUS = H / 'state/sessions/bus.jsonl'
T_URL, T_GIT = 15, 10

HEAD_RE = re.compile(r'^\s*(?:#+\s*)?(?:\*\*)?(Приёмка|Приемка|Критерии приёмки|Критерии приемки|Условия приёмки|Acceptance(?: criteria)?|Готово,? когда)(?:\*\*)?\s*:?\s*(?:\*\*)?', re.I | re.M)
URL_RE = re.compile(r'https?://[^\s)»"\'<>,;]+')
SHA_RE = re.compile(r'\b[0-9a-f]{7,40}\b')
PATH_RE = re.compile(r'(?<![\w/.-])((?:/(?:[\w.-]+/)*|(?:[\w.-]+/)+)[\w.-]+\.[A-Za-z0-9]{1,6})(?![\w/])')
NUM_RE = re.compile(r'(=|==|≤|<=|≥|>=|<|>)\s*(\d+(?:[.,]\d+)?)')
CMD_RE = re.compile(r'\b[\w.-]+\.(?:py|sh)\b')
IP_RE = re.compile(r'\b\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}\b')
LINES_RE = re.compile(r'(?:≤|<=|не более)\s*(\d+)\s*строк', re.I)
VAL_RE = r'(?<![\w.:/-])(\d+(?:[.,]\d+)?)(?![.:/]\d|[\w-])'


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


def tz_text(raw: str) -> str:
    """tz — путь к файлу ТЗ (от корня дома) или сам текст задачи."""
    if raw and '\n' not in raw and len(raw) < 300:
        f = H / raw
        if f.is_file():
            return f.read_text(encoding='utf-8', errors='replace')
    return raw or ''


def block(text: str) -> tuple[str | None, str]:
    """Блок приёмки: строка-заголовок (синонимы) и всё до пустой строки/следующего заголовка."""
    m = HEAD_RE.search(text or '')
    if not m:
        return None, ''
    out = []
    for i, l in enumerate(text[m.end():].splitlines()):
        if i and (not l.strip() or re.match(r'^\s*#', l)):
            break
        out.append(l)
    return m.group(1), '\n'.join(out).strip()


def items(blk: str) -> list[str]:
    """Пункты: строки-списки, иначе — по «;» и концу предложения."""
    lines = [re.sub(r'^\s*(?:[-*•]|\d+[.)])\s*', '', l).strip() for l in blk.splitlines() if l.strip()]
    if len(lines) == 1:
        lines = [x.strip() for x in re.split(r';\s+|(?<=[.!])\s+(?=[А-ЯA-Z«])', lines[0]) if x.strip()]
    return [l.rstrip('.;') for l in lines if len(l) > 3]


def kind(it: str) -> str:
    if URL_RE.search(it): return 'url'
    if re.search(r'коммит|\bcommit\b', it): return 'commit'
    if LINES_RE.search(it) and re.search(r'сдач|письм', it, re.I): return 'lines'
    if NUM_RE.search(it) and not IP_RE.search(it): return 'number'
    if PATH_RE.search(it) and not CMD_RE.search(it.split()[0] if it.split() else ''): return 'file'
    return 'judge'


def gate(raw: str) -> tuple[bool, str]:
    head, blk = block(tz_text(raw))
    if head is None:
        return False, 'ТЗ без блока «Приёмка:» — раздачи нет. Допиши «Приёмка: <URL отдаёт 200 | файл <путь> есть | коммит | число с порогом>»'
    ks = [kind(i) for i in items(blk)]
    if not any(k != 'judge' for k in ks):
        return False, f'блок «{head}» есть, но в нём нет проверяемого условия (URL, путь файла, коммит, число с порогом)'
    return True, f'приёмка: {len(ks)} пунктов, проверяемых без LLM {sum(k != "judge" for k in ks)}'


def http_code(url: str):
    try:
        return urllib.request.urlopen(urllib.request.Request(url, headers={'User-Agent': 'home-kit-accept/1'}), timeout=T_URL).status
    except urllib.error.HTTPError as e:
        return e.code
    except Exception as e:
        return type(e).__name__


def _jsonl(p: pathlib.Path) -> list[dict]:
    out = []
    if p.exists():
        for l in p.read_text(encoding='utf-8').splitlines():
            try:
                out.append(json.loads(l))
            except Exception:
                pass
    return out


def evidence(label: str) -> dict:
    done = [r.get('text', '') for r in _jsonl(TURNS) if r.get('session') == label and r.get('kind') == 'done']
    letters = [r.get('text', '') for r in _jsonl(BUS) if r.get('kind') == 'msg' and r.get('session') == label]
    return {'done': '\n'.join(done), 'letter': '\n---\n'.join(letters) if letters else 'письма на шине нет'}


def check(it: str, ev: dict):
    """→ (True/False | None — судье, доказательство)."""
    k = kind(it)
    blob = ev['done'] + '\n' + ev['letter']
    if k == 'url':
        url = URL_RE.search(it).group(0)
        m = re.search(r'\b([1-5]\d\d)\b', it.replace(url, ''))
        want = int(m.group(1)) if m else 200
        c = http_code(url)
        return c == want, f'{url} → {c} (ждали {want})'
    if k == 'file':
        bad, ps = [], PATH_RE.findall(it)
        for p in ps[:4]:
            f = pathlib.Path(p) if p.startswith('/') else H / p
            sz = f.stat().st_size if f.exists() else -1
            if sz <= 0:
                bad.append(f'{p}: ' + ('пуст' if sz == 0 else 'нет'))
        return (not bad), ('; '.join(bad) if bad else ', '.join(f'{p} есть' for p in ps[:4]))
    if k == 'commit':
        for s in SHA_RE.findall(it + ' ' + blob)[:6]:
            r = subprocess.run(['git', '-C', str(H), 'cat-file', '-t', s], capture_output=True, text=True, timeout=T_GIT)
            if r.stdout.strip() == 'commit':
                return True, f'коммит {s} есть'
        return False, 'хеша коммита в сдаче нет'
    if k == 'lines':
        n = int(LINES_RE.search(it).group(1))
        if ev['letter'].startswith('письма на шине нет'):
            return False, 'письма на шине нет'
        ls = [sum(1 for l in m.splitlines() if l.strip()) for m in ev['letter'].split('\n---\n')]
        return max(ls) <= n, f'письмо координатору: {max(ls)} непустых строк (условие ≤ {n})'
    if k == 'number':
        op, val = NUM_RE.search(it).groups()
        val = float(val.replace(',', '.'))
        head = re.sub(r'\S*[/\d]\S*|\S+-\S+', ' ', it[:NUM_RE.search(it).start()])
        words = re.findall(r'[\wа-яё]{3,}', head.lower())[-2:]
        cmp = lambda x: {'=': x == val, '==': x == val, '≤': x <= val, '<=': x <= val, '≥': x >= val, '>=': x >= val, '<': x < val, '>': x > val}[op]
        low = blob.lower()
        for w in reversed(words):
            vals = [m.group(1) for m in re.finditer(re.escape(w[:5]) + r'[\wа-яё«»"\s:.—=≤≥<>→-]{0,40}?' + VAL_RE, low)]
            if vals:
                v = float(vals[0].replace(',', '.'))
                return cmp(v), f'в сдаче «{w[:5]}…» = {vals[0]} (условие {op} {val:g})'
        return None, 'числа в сдаче не нашёл'
    return None, 'судье'


def cmd_check(label: str) -> int:
    add = [r for r in _jsonl(REG) if r.get('label') == label and 'ev' not in r]
    if not add:
        print(f'{label}: в реестре нет'); return 2
    head, blk = block(tz_text(add[-1].get('tz', '')))
    if head is None:
        print(f'{label}: в ТЗ нет блока приёмки'); return 2
    ev = evidence(label)
    res = []
    for it in items(blk):
        ok, why = check(it, ev)
        res.append({'p': it, 'ok': ok, 'evidence': why, 'by': 'zero-llm' if ok is not None else 'judge'})
    verdict = 'return' if any(r['ok'] is False for r in res) else ('judge' if any(r['ok'] is None for r in res) else 'ok')
    with REG.open('a', encoding='utf-8') as f:
        f.write(json.dumps({'ev': 'accept', 'label': label, 'verdict': verdict, 'items': res, 'at': iso()}, ensure_ascii=False) + '\n')
    for r in res:
        print(f"  {'✓' if r['ok'] else ('✗' if r['ok'] is False else '?')} {r['p'][:70]} — {r['evidence']}")
    print(f'{label}: {verdict}')
    return 0 if verdict == 'ok' else 1


def selftest() -> int:
    t = 'Сделать X.\nПриёмка: файл out/x.md есть; коммит; число строк ≤ 5.\n'
    h, b = block(t)
    assert h == 'Приёмка' and len(items(b)) == 3, (h, items(b))
    assert [kind(i) for i in items(b)] == ['file', 'commit', 'number'], [kind(i) for i in items(b)]
    assert gate(t)[0] and not gate('без блока')[0]
    print('selftest ok'); return 0


if __name__ == '__main__':
    a = sys.argv[1:]
    if a[:1] == ['gate'] and len(a) > 1:
        ok, msg = gate(a[1]); print(msg); sys.exit(0 if ok else 1)
    if a[:1] == ['check'] and len(a) > 1:
        sys.exit(cmd_check(a[1]))
    if a[:1] == ['selftest']:
        sys.exit(selftest())
    print(__doc__)