гейт кода в закрытии задачи: правка харнеса = коммит + тест + зелёный тест

Кейс 2026-10-18 (t-261018103940): Коди закрыл задачу итогом «x» до конца
правки tg_inbox, без теста и без коммита. Проверка — в complete (не в промпте):
- незакоммиченные правки harness/ cli.py tests/ scripts/, сделанные тиком
  задачи после её клейма → отказ со списком файлов. «Сделанные тиком» — по
  трассе: edit/write по пути, shell — путь + признак записи (sed -i, >, tee,
  cp/mv/rm, patch, git apply, open(…,'w')); чтение (grep, sed -n) и запуск
  (python cli.py …) — не правка; нет трассы — по времени изменения файла.
  Иначе чужая незакоммиченная правка того же времени блокировала бы задачу;
- коммиты после клейма (точно по %ct, не по секунде --since) меняют код без
  tests/test_*.py → отказ; исключение — complete --no-test «причина»
  (причина уходит Alex в итоге);
- изменённые тесты запускаются при закрытии, FAIL/падение → отказ;
- в итог Alex автоматически: «Код: <хеши> · тесты PASS: …»;
- правило — в постановке батча задач; SIMBIOT_CODE_GATE=off выключает.
Сухой прогон на задаче Коди: сейчас — «Код: 70a2015 · тесты PASS:
tests/test_tg_group_ux.py»; в 10:42 был бы отказ (tg_inbox.py не закоммичен).
Тест: test_code_gate 14 PASS; полный набор — без регрессий.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

cli.py

@@ -53,6 +53,8 @@ def main():
     p_cmp = sub.add_parser("complete", help="закрыть running-задачу с результатом (батч-тик, t-260930101325)")
     p_cmp.add_argument("task_id")
     p_cmp.add_argument("-a", "--answer", required=True, help="результат задачи")
+    p_cmp.add_argument("--no-test", default=None, metavar="ПРИЧИНА",
+                       help="правка кода без теста — только с причиной (уйдёт Alex вместе с итогом)")
 
     p_pr = sub.add_parser("promote", help="промоушен ready-задачи на уровень выше (S→M→L)")
     p_pr.add_argument("task_id")
@@ -299,8 +301,12 @@ def main():
         # батч-тик (t-260930101325): исполнитель закрывает задачи пачки
         # по отдельности; с await_delivery результат уходит в TG (идемпотентно)
         es.init_db()
+        try:
             tid, delivered = commands.complete_task_with_delivery(
-            args.task_id, args.answer, level="manual")
+                args.task_id, args.answer, level="manual", no_test_reason=args.no_test)
+        except ValueError as e:
+            print(f"ОТКАЗ: {e}")
+            sys.exit(2)
         print(f"OK: задача {tid} завершена"
               + (" (доставлено в TG)" if delivered else ""))
     elif args.cmd == "close":

harness/commands.py

@@ -262,9 +262,151 @@ def looks_like_budget_stub(answer: str) -> bool:
 
 MIN_ANSWER = 10  # символов в итоге задачи: короче — не результат («x», «готово»)
 
+# ────────────── гейт кода: правка харнеса = коммит + тест + зелёный тест ──────────────
+# Кейс 2026-10-18 (t-261018103940): Коди сам доработал tg_inbox, но закрыл
+# задачу итогом «x» ДО конца правки, без теста и без коммита. Проверка — в
+# закрытии задачи (не просьба в промпте): изменения кода, сделанные после
+# клейма задачи, обязаны быть закоммичены, код — с тестом, тест — зелёным.
+CODE_PATHS = ("harness", "cli.py", "tests", "scripts")
+_REPO_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+
+
+def _git(repo: str, *args) -> str:
+    import subprocess
+    r = subprocess.run(["git", "-C", repo, *args], capture_output=True, text=True, timeout=30)
+    return r.stdout if r.returncode == 0 else ""
+
+
+def _run_test(repo: str, path: str) -> tuple:
+    """(ok, хвост вывода) — тест-скрипт харнеса; FAIL-строки или падение = не ок."""
+    import subprocess
+    import sys as _sys
+    py = os.path.join(repo, "venv", "bin", "python")
+    try:
+        r = subprocess.run([py if os.path.exists(py) else _sys.executable, path], cwd=repo,
+                           capture_output=True, text=True, timeout=240)
+    except subprocess.TimeoutExpired:
+        return False, "таймаут 240 с"
+    out = (r.stdout or "") + (r.stderr or "")
+    fails = [ln for ln in out.splitlines() if ln.startswith("FAIL ")]
+    bad = bool(fails) or "Traceback (most recent call last)" in out or r.returncode != 0
+    return (not bad), "\n".join((fails or out.splitlines())[-3:])[:300]
+
+
+_WRITE_RE = re.compile(r"sed\s+(?:-\S+\s+)*-i|perl\s+-\S*i|>>?|\btee\b|\b(?:cp|mv|rm|install|touch)\s|"
+                       r"\bpatch\b|git\s+(?:apply|checkout|restore|am)\b|open\([^)]*['\"][wa]")
+_NOISE_RE = re.compile(r"\d?>&\d|\d?>>?\s*/dev/null")
+
+
+def _touched_by_trace(task: dict):
+    """Правил ли тик задачи файл: предикат path → bool или None (трассы нет).
+
+    edit/write — по пути во входе; shell — путь в команде И признак записи в ней
+    (sed -i, >, tee, cp/mv/rm, patch, git apply, open(…,'w')). Чтение (grep, cat,
+    sed -n) и запуск (python cli.py …) — не правка: кейс 18.10 — «cli.py» есть
+    в каждой команде тика, а чужая правка cli.py блокировала бы закрытие."""
+    sess = task.get("session_id") or ""
+    if not sess.startswith("s-"):
+        return None
+    try:
+        db = es._conn()
+        try:
+            rows = db.execute("SELECT data FROM events WHERE event_type='ToolCalled' "
+                              "AND json_extract(data,'$.trace_id')=?", (sess[2:],)).fetchall()
+        finally:
+            db.close()
+    except Exception:
+        return None
+    if not rows:
+        return None
+    import json as _json
+    edits, writes = [], []
+    for (d,) in rows:
+        d = _json.loads(d)
+        inp = d.get("input") or ""
+        try:
+            obj = _json.loads(inp)
+        except ValueError:
+            obj = {}
+        if d.get("tool") in ("edit", "write"):
+            edits.append(str(obj.get("path") or obj.get("file_path") or inp))
+        elif d.get("tool") == "shell":
+            cmd = str(obj.get("command") or inp)
+            if _WRITE_RE.search(_NOISE_RE.sub(" ", cmd)):
+                writes.append(cmd)
+
+    def touched(path: str) -> bool:
+        base = os.path.basename(path)
+        if any(e.endswith(path) or e.endswith("/" + base) or e == base for e in edits):
+            return True
+        return any(path in c or re.search(rf"(?<![\w./-]){re.escape(base)}\b", c) for c in writes)
+    return touched
+
+
+def code_change_check(task: dict, no_test_reason: str = None):
+    """(ok, причина отказа, сводка для итога) по правкам кода во время задачи."""
+    repo = os.environ.get("SIMBIOT_CODE_REPO", "").strip() or _REPO_ROOT
+    since = task.get("claimed_at")
+    if (os.environ.get("SIMBIOT_CODE_GATE", "on").lower() == "off" or not since
+            or not os.path.isdir(os.path.join(repo, ".git"))):
+        return True, "", ""
+    since_ts = datetime.fromisoformat(since).timestamp()
+    # что тик задачи реально трогал: входы его инструментов (trace = session без «s-»).
+    # Без этого чужая незакоммиченная правка того же времени (кейс 18.10:
+    # scripts/deploy-claude-code.sh) блокировала бы закрытие навсегда.
+    touched = _touched_by_trace(task)
+    dirty = []
+    for ln in _git(repo, "status", "--porcelain", "--", *CODE_PATHS).splitlines():
+        path = ln[3:].strip().split(" -> ")[-1].strip('"')
+        full = os.path.join(repo, path)
+        if touched is not None and not touched(path):
+            continue  # тик этот файл не правил — чужая правка
+        try:
+            if os.path.getmtime(full) >= since_ts - 5:
+                dirty.append(path)
+        except OSError:
+            dirty.append(path)  # удалён после клейма — тоже правка
+    if dirty:
+        return False, ("в коде есть незакоммиченные изменения, сделанные в этой задаче: " + ", ".join(dirty[:8])
+                       + " — закоммить их (git add <файлы> && git commit) вместе с тестом или "
+                         "откати; потом закрывай задачу"), ""
+    # --since у git — с точностью до секунды: коммит той же секунды, но ДО клейма
+    # попадал в задачу; точный фильтр — по времени коммита (%ct)
+    log = _git(repo, "log", f"--since={since}", "--format=@%h %ct", "--name-only", "--", *CODE_PATHS)
+    commits, files, take = [], set(), False
+    for ln in log.splitlines():
+        if ln.startswith("@"):
+            h, ct = ln[1:].split()
+            take = int(ct) > int(since_ts)  # секунда клейма — ещё не работа задачи
+            if take:
+                commits.append(h)
+        elif ln.strip() and take:
+            files.add(ln.strip())
+    if not commits:
+        return True, "", ""
+    code = sorted(f for f in files if f.endswith(".py") and (f.startswith("harness/") or f == "cli.py"))
+    tests = sorted(f for f in files if re.match(r"tests/test_\w+\.py$", f)
+                   and os.path.exists(os.path.join(repo, f)))
+    if code and not tests and not no_test_reason:
+        return False, ("изменён код (" + ", ".join(code[:6]) + ") без теста: добавь/обнови "
+                       "tests/test_*.py, закоммить и закрывай снова; если тест невозможен — "
+                       "complete --no-test \"причина\" (причина уйдёт Alex)"), ""
+    results = []
+    for tf in tests:
+        ok, tail = _run_test(repo, tf)
+        if not ok:
+            return False, f"тест {tf} не проходит: {tail} — почини и закрывай снова", ""
+        results.append(tf)
+    note = "Код: " + ", ".join(commits[:6])
+    if results:
+        note += " · тесты PASS: " + ", ".join(results[:4])
+    if no_test_reason and code and not tests:
+        note += f" · без теста: {no_test_reason[:200]}"
+    return True, "", note
+
 
 def complete_task_with_delivery(task_id: str, answer: str, level: str,
-                                 result_ref: str = None):
+                                 result_ref: str = None, no_test_reason: str = None):
     """TaskCompleted + немедленная доставка результата в TG-чат.
 
     Используется исполнителем (часовой/дневной тик) как атомарная замена
@@ -294,6 +436,11 @@ def complete_task_with_delivery(task_id: str, answer: str, level: str,
             f"задача {task_id} ПЕРЕВОДЖЕНА в ready: отпиской «бюджет исчерпан» "
             f"задачу закрывать нельзя. Либо доделай и завершай с реальным "
             f"результатом, либо явно requeue/failed.")
+    ok, why, note = code_change_check(t, no_test_reason)
+    if not ok:
+        raise ValueError(f"задача {task_id} не закрыта: {why}")
+    if note:
+        answer = f"{answer.rstrip()}\n\n{note}"
     # TaskCompleted — независимо от доставки (событие-истина о завершении)
     es.append_event("task", task_id, "TaskCompleted", {
         "result_ref": result_ref, "answer": answer[:2000],

harness/scheduler.py

@@ -659,6 +659,9 @@ def build_context(level_cfg, model: dict, batch=None):
             "по порядку списка — первая самая важная; не растягивай: тик ограничен таймаутом и "
             f"бюджетом шагов; суммарная оценка ~{sum(_est_minutes(t) for t in batch)}м):\n"
             + "\n".join(lines)
+            + "\nПравишь код харнеса (harness/, cli.py, tests/, scripts/) — задача закрывается "
+              "только после git commit с тестом tests/test_*.py, тест должен проходить: "
+              "complete это проверит и откажет; тест невозможен — complete --no-test \"причина\"."
             + "\nКаждую завершённую закрывай ОТДЕЛЬНО сразу после выполнения: "
               "./venv/bin/python cli.py complete <task_id> -a \"результат\" "
               "(идемпотентно: завершённая уже не завершится повторно; с флагом "

tests/test_code_gate.py

@@ -0,0 +1,155 @@
+# -*- coding: utf-8 -*-
+"""Гейт кода в закрытии задачи (commands.code_change_check) — hermetic, временный git-репо.
+
+Запуск: ./venv/bin/python tests/test_code_gate.py
+
+Кейс 2026-10-18: Коди закрыл задачу итогом «x» до конца правки tg_inbox, без
+теста и коммита. Проверяет: задача без правок кода — закрывается как раньше;
+незакоммиченная правка после клейма — отказ; старая чужая грязь (до клейма) —
+не мешает; код без теста — отказ, с --no-test «причина» — можно, причина в
+итоге; код + зелёный тест — закрыта, хеш и тест в итоге; красный тест — отказ;
+SIMBIOT_CODE_GATE=off — гейт выключен.
+"""
+import os
+import subprocess
+import sys
+import tempfile
+import time
+
+sys.path.insert(0, "/root/simbiot-harness")
+_tmp = tempfile.mkdtemp(prefix="simbiot_codegate_")
+os.environ["SIMBIOT_DB"] = os.path.join(_tmp, "harness.db")
+REPO = os.path.join(_tmp, "repo")
+os.environ["SIMBIOT_CODE_REPO"] = REPO
+os.environ.pop("SIMBIOT_CODE_GATE", None)
+
+from harness import commands, es
+
+es.init_db()
+fails = []
+
+
+def check(name, cond, extra=""):
+    print(("PASS" if cond else "FAIL"), name, ("| " + str(extra)) if not cond else "")
+    if not cond:
+        fails.append(name)
+
+
+def git(*a):
+    subprocess.run(["git", "-C", REPO, "-c", "user.name=t", "-c", "user.email=t@t", *a],
+                   check=True, capture_output=True)
+
+
+def write(rel, text, mtime=None):
+    p = os.path.join(REPO, rel)
+    os.makedirs(os.path.dirname(p), exist_ok=True)
+    open(p, "w").write(text)
+    if mtime:
+        os.utime(p, (mtime, mtime))
+
+
+os.makedirs(REPO)
+git("init", "-q")
+write("harness/a.py", "x = 1\n")
+write("scripts/old.sh", "echo old\n")
+git("add", ".")
+git("commit", "-q", "-m", "init")
+write("scripts/old.sh", "echo old dirty\n", mtime=time.time() - 3600)  # чужая старая грязь
+N = [0]
+
+
+def task():
+    N[0] += 1
+    tid = f"t-{N[0]}"
+    es.append_event("task", tid, "TaskSubmitted", {"id": tid, "title": "правка", "size": "S"})
+    es.append_event("task", tid, "TaskClaimed", {"level": "min"})
+    time.sleep(1.1)  # git --since с точностью до секунды
+    return tid
+
+
+def done(tid, answer="сделана правка харнеса", **kw):
+    try:
+        commands.complete_task_with_delivery(tid, answer, level="min", **kw)
+        return True, ""
+    except ValueError as e:
+        return False, str(e)
+
+
+def answer_of(tid):
+    import json
+    return [json.loads(e["data"])["answer"] for e in es.get_events(0, 10**6)
+            if e["event_type"] == "TaskCompleted" and e["aggregate_id"] == tid][-1]
+
+
+t = task()
+ok, why = done(t, "ответил в чат, код не трогал")
+check("без правок кода — закрывается как раньше, итог без приписки",
+      ok and "Код:" not in answer_of(t), why)
+check("старая чужая грязь (до клейма) не мешает", ok)
+
+t = task()
+write("harness/a.py", "x = 2\n")
+ok, why = done(t)
+check("незакоммиченная правка — отказ с файлом", not ok and "harness/a.py" in why, why)
+git("add", "harness/a.py")
+git("commit", "-q", "-m", "a=2")
+ok, why = done(t)
+check("код без теста — отказ", not ok and "без теста" in why, why)
+ok, why = done(t, no_test_reason="правка константы, тест невозможен")
+check("--no-test с причиной — закрыта, причина в итоге",
+      ok and "без теста: правка константы" in answer_of(t), why)
+
+t = task()
+write("harness/a.py", "x = 3\n")
+write("tests/test_a.py", "print('PASS a ok')\n")
+git("add", ".")
+git("commit", "-q", "-m", "a=3 + test")
+ok, why = done(t)
+ans = answer_of(t) if ok else ""
+check("код + зелёный тест — закрыта", ok, why)
+check("в итоге хеш коммита и тест", "Код: " in ans and "тесты PASS: tests/test_a.py" in ans, ans)
+
+t = task()
+write("harness/a.py", "x = 4\n")
+write("tests/test_a.py", "print('FAIL a broken')\nraise SystemExit(1)\n")
+git("add", ".")
+git("commit", "-q", "-m", "a=4 broken test")
+ok, why = done(t)
+check("красный тест — отказ", not ok and "не проходит" in why, why)
+
+# чужая незакоммиченная правка того же времени: тик её не трогал — не блокирует
+N[0] += 1
+tid = f"t-{N[0]}"
+es.append_event("task", tid, "TaskSubmitted", {"id": tid, "title": "правка", "size": "S"})
+es.append_event("task", tid, "TaskClaimed", {"level": "min", "session_id": "s-min-777"})
+time.sleep(1.1)
+es.append_event("trace", "min-777", "ToolCalled", {"trace_id": "min-777", "tool": "edit",
+                "input": '{"path": "/x/repo/harness/b.py"}'})
+es.append_event("trace", "min-777", "ToolCalled", {"trace_id": "min-777", "tool": "shell",
+                "input": '{"command": "./venv/bin/python cli.py tg-reply m-1 \\"ok\\""}'})
+es.append_event("trace", "min-777", "ToolCalled", {"trace_id": "min-777", "tool": "shell",
+                "input": '{"command": "grep -n tg_reply cli.py 2>/dev/null | head"}'})
+es.append_event("trace", "min-777", "ToolCalled", {"trace_id": "min-777", "tool": "shell",
+                "input": '{"command": "sed -i s/a/b/ scripts/mine.sh"}'})
+write("cli.py", "# чужая правка cli\n")    # запускали и читали, но не правили — не своё
+write("scripts/mine.sh", "echo b\n")        # правил sed -i — своё
+write("harness/b.py", "y = 1\n")          # своя правка (в трассе)
+write("scripts/other.sh", "echo чужое\n")  # чужая правка (не в трассе)
+ok, why = done(tid)
+check("своя незакоммиченная правка (в трассе) — отказ", not ok and "harness/b.py" in why, why)
+check("чужая правка того же времени не в списке", "scripts/other.sh" not in why, why)
+check("запуск и grep cli.py — не правка cli.py", "cli.py" not in why, why)
+check("sed -i в shell — своя правка", "scripts/mine.sh" in why, why)
+git("add", "harness/b.py", "scripts/mine.sh")
+write("tests/test_a.py", "print('PASS b ok')\n")
+git("add", "tests/test_a.py")
+git("commit", "-q", "-m", "b + test")
+ok, why = done(tid)
+check("свою закоммитил — закрыта, чужая грязь не мешает", ok, why)
+
+os.environ["SIMBIOT_CODE_GATE"] = "off"
+ok, why = done(t)
+check("SIMBIOT_CODE_GATE=off — гейт выключен", ok, why)
+
+print(f"\n{'OK' if not fails else 'FAILED'}: {len(fails)} fail(s)")
+sys.exit(1 if fails else 0)