107 KiB
107 KiB
Мост Jira — агент (jira-agent) — план реализации
For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (
- [ ]) syntax for tracking.
Goal: Один файл agent.py для виртуалки с доступом к Jira: опрашивает хаб, выполняет команды «проверь, потом сделай», ведёт журнал на диске, возвращает хабу состояние задач.
Architecture: src/agent.py (Python 3.9, только stdlib): JiraClient (REST v2 + Agile, 429/503 по Retry-After), HubClient, Journal (jsonl, только дозапись), Executor (по методу на тип команды, каждая — чтение → сравнение → запись разницы), Runner (цикл claim → пауза → журнал → выполнение → отчёт до подтверждения; восстановление по журналу при старте). Политику темпа агент не знает: пауза приходит с командой, а зазор внутри многошаговых команд — из настроек хаба.
Tech Stack: Python 3.9.6 (/usr/bin/python3 и здесь, и на виртуалке), stdlib urllib, json, http.server (только в тестах), unittest.
Spec: ../jira-hub/openspec/changes/bridge-core/ — specs/agent/spec.md (агент), specs/commands/spec.md (контракт команд), specs/hub-api/spec.md (маршруты /agent/*), design.md. План хаба — ../jira-hub/docs/superpowers/plans/2026-10-11-bridge-core-hub.md.
Global Constraints
- Python
>= 3.9, только стандартная библиотека; никакихmatch,X | Yв аннотациях,str.removeprefixдопустим (3.9). Тесты запускать/usr/bin/python3 -m unittest discover -s tests -t .. - Учётные данные Jira — только в
agent.json; в хаб, журнал и логи не попадают. - Одна команда в полёте;
started/finished/ackedвjournal.jsonl; отчёт повторяется до подтверждения хаба. - Все вызовы Jira идут через
JiraClient.call: счётчикиcalls,rate_limited,last_call_at; 429/503 —Retry-Afterили 60 с × попытка, до 5 раз. - Семантика связей:
sourceблокируетtarget;POST /issueLink—inwardIssue = source,outwardIssue = target; в GET у XoutwardIssue = Yзначит «X → Y». - Язык кода и комментариев — русский, идентификаторы — английские; коммиты — русские, с
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>. - Файл
agent.pyлежит вsrc/(хук require-plan защищаетsrc/); на виртуалку копируется один файл плюсagent.json.
Review Focus
- Описание в
ensure_fieldsс переводами строк и wiki-разметкой должно сравниваться точно, без «нормализации» (иначе каждая команда будет перезаписывать описание). Тест — Task 3 (test_fields_description_exact_compare). ensure_status, когда текущий статус уже внутри пути (например «Разработка» при пути Разработка → Готово): выполняется только остаток пути. Тест — Task 4 (test_status_resumes_from_middle).refreshпри пустом спискеkeys(все временные ключи отвалились) —doneс нулём задач, без вызовов Jira. Тест — Task 5 (test_refresh_empty_keys).- Отчёт хабу, когда хаб вернул 404/422 (команда отменена с
forceили удалена): агент помечаетackedи идёт дальше, не зацикливается. Тест — Task 6 (test_result_rejected_is_acked). - Ctrl-C во время паузы перед командой: команда не начинается, журнал без
started, процесс выходит. Тест — Task 6 (test_stop_during_pause).
Структура файлов
| Файл | Ответственность |
|---|---|
src/agent.py |
весь агент: конфиг, HTTP-клиенты, журнал, исполнители, чтение состояния, цикл, CLI (run, check, prune) |
src/agent.example.json |
шаблон настроек |
tests/_common.py |
путь к src, общие настройки хаба для тестов, фабрики |
tests/fake_jira.py |
фейковый Jira Server в памяти (http.server) |
tests/fake_hub.py |
фейковый хаб (маршруты /agent/*) |
tests/test_*.py |
unittest |
README.md, CLAUDE.md, ARCHITECTURE.md |
доставка, запуск, launchd, prune |
Task 1: Скелет агента, HTTP-клиенты, фейковый Jira, check
Files:
- Create:
src/agent.py,src/agent.example.json,tests/__init__.py(пустой),tests/_common.py,tests/fake_jira.py - Test:
tests/test_clients.py
Interfaces:
-
Produces (в
agent.py):AGENT_VERSION = 1,now_iso() -> str,log(msg),load_config(path) -> dict,HttpError(code, text, headers),Http(base, headers, timeout, ctx).call(method, path, body=None, params=None),JiraClient(cfg).get/post/put/delete, счётчикиcalls,rate_limited,last_call_at,reset_counters(),sleep(подменяемый),HubClient(cfg).claim(agent_version, settings_version, last_jira_call_at),.settings(),.command(id),.result(id, payload),.state(payload),.known_keys(),main(argv) -> intс подкомандамиrun | check | prune(run и prune — заглушки до Task 6/2). -
Produces (тесты):
fake_jira.FakeJira()(.base_url,.add(key, in_base=True, **fields),.issues,.links,.comments,.sprints,.requests,.writes(),.fail_next,.stop()),_common.SETTINGS,_common.jira_cfg(fake) -> dict,_common.agent(модуль). -
Step 1: Общие тестовые файлы
tests/_common.py:
"""Общее для тестов: путь к src, настройки хаба как их отдаёт /agent/settings, конфиг агента под фейковый Jira."""
import os
import sys
SRC = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "src")
if SRC not in sys.path:
sys.path.insert(0, SRC)
import agent # noqa: E402
ZERO_PACING = {"ensure_issue": 0, "ensure_fields": 0, "ensure_comment": 0, "ensure_status": 0, "ensure_link": 0,
"ensure_sprint": 0, "refresh": 0}
SETTINGS = {
"base_jql": "project = AURORA AND labels = AWPCA",
"project_key": "AURORA",
"issue_types": {"task": {"name": "Задача", "id": "10100"}, "epic": {"name": "Фича", "id": "12300"}},
"epic_name_field_id": "customfield_15304",
"field_ids": {"description": "description", "priority": "priority", "duedate": "duedate", "labels": "labels",
"planned_start": "customfield_10801", "planned_end": "customfield_10802"},
"epic_link": {"mode": "field", "field_id": "customfield_15302"},
"priority_map": {"Наивысший": "Критичный", "Высокий": "Высокий", "Средний": "Средний", "Низкий": "Низкий"},
"priority_value_key": "name",
"labels": ["AWPCA"],
"status_labels": {},
"kind_labels": {},
"comment_marker": "Актуализация 10.10.2026",
"transitions_map": {},
"link_types": {"блокирует начало": "ASUPR_Блокер_Начала", "связанные": "ASUPR_Связанные"},
"board_id": 4216,
"pacing": dict(ZERO_PACING),
"min_agent_version": 1,
"settings_version": 1,
}
def jira_cfg(fake, **over):
cfg = {"hub_url": "http://127.0.0.1:1", "hub_token": "agent-token",
"jira": {"base_url": fake.base_url, "auth": {"type": "basic", "user": "me", "password": "secret"}},
"verify_ssl": True, "ca_bundle": None, "timeout": 5, "poll_seconds": 0}
cfg.update(over)
return cfg
def no_sleep(_seconds):
pass
tests/fake_jira.py:
"""Фейковый Jira Server для тестов агента: состояние в памяти, http.server в потоке,
только маршруты, которые использует агент. Семантика связей — как у настоящего Jira Server."""
import json
import re
import threading
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from urllib.parse import parse_qs, urlparse
WORKFLOW = { # статус → [(имя перехода, целевой статус)]
"Ожидание разработки": [("Взять в разработку", "Разработка"), ("Ожидает анализа", "Ожидает анализа"), ("Снято", "Снято")],
"Разработка": [("Малое исправление", "Готово к тестированию"), ("Готово", "Готово"), ("Снято", "Снято")],
"Ожидает анализа": [("Анализ", "Анализ"), ("Снято", "Снято")],
"Анализ": [("Готово", "Готово")],
"Готово к тестированию": [("Тестирование", "Тестирование")],
"Готово": [],
"Снято": [],
}
LINK_TYPES = [{"name": "ASUPR_Блокер_Начала", "inward": "Начало заблокировано", "outward": "Блокирует начало"},
{"name": "ASUPR_Связанные", "inward": "Связанные", "outward": "Связанные"}]
CUSTOM = ("customfield_10801", "customfield_10802", "customfield_15302", "customfield_15304")
class FakeJira:
def __init__(self, project="AURORA"):
self.project = project
self.issues = {}
self.comments = {}
self.links = []
self.sprints = [{"id": 1, "name": "Спринт 1", "state": "active"}, {"id": 2, "name": "Спринт 2", "state": "future"}]
self.seq, self.link_seq, self.comment_seq = 1000, 100, 10
self.fail_next = [] # [(код, {заголовки})] — следующие ответы с ошибкой
self.requests = [] # [(метод, путь)]
self.lock = threading.Lock()
fake = self
class Handler(BaseHTTPRequestHandler):
def log_message(self, *a):
pass
def do_GET(self):
fake.handle(self, "GET")
def do_POST(self):
fake.handle(self, "POST")
def do_PUT(self):
fake.handle(self, "PUT")
def do_DELETE(self):
fake.handle(self, "DELETE")
self.server = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
threading.Thread(target=self.server.serve_forever, daemon=True).start()
@property
def base_url(self):
return "http://127.0.0.1:%d" % self.server.server_port
def stop(self):
self.server.shutdown()
def add(self, key, in_base=True, **fields):
base = {"summary": "Тема " + key, "description": None, "status": "Ожидание разработки", "priority": "Средний",
"assignee": None, "reporter": "someone", "duedate": None, "labels": ["AWPCA"] if in_base else [],
"issuetype": "Задача", "created": "2026-10-01T10:00:00.000+0300", "updated": "2026-10-01T10:00:00.000+0300",
"sprint": None}
base.update({c: None for c in CUSTOM})
base.update(fields)
self.issues[key] = base
return base
def writes(self):
return [r for r in self.requests if r[0] in ("POST", "PUT", "DELETE")]
# --- представление ---------------------------------------------------------
def _user(self, name):
return {"name": name, "displayName": name} if name else None
def issue_json(self, key):
f = self.issues[key]
fields = {"summary": f["summary"], "description": f["description"], "status": {"name": f["status"]},
"priority": {"name": f["priority"]}, "assignee": self._user(f["assignee"]), "reporter": self._user(f["reporter"]),
"duedate": f["duedate"], "labels": list(f["labels"]), "issuetype": {"name": f["issuetype"]},
"created": f["created"], "updated": f["updated"], "issuelinks": self.links_of(key)}
fields.update({c: f.get(c) for c in CUSTOM})
return {"key": key, "id": str(10000 + sorted(self.issues).index(key)), "fields": fields}
def links_of(self, key):
out = []
for l in self.links:
if l["source"] == key:
out.append({"id": str(l["id"]), "type": {"name": l["type"]}, "outwardIssue": {"key": l["target"]}})
elif l["target"] == key:
out.append({"id": str(l["id"]), "type": {"name": l["type"]}, "inwardIssue": {"key": l["source"]}})
return out
def search_keys(self, jql):
m = re.search(r"key in \(([^)]*)\)", jql or "")
if m:
keys = [k.strip() for k in m.group(1).split(",")]
return [k for k in keys if k in self.issues]
if "reporter = currentUser()" in (jql or ""):
return [k for k, f in self.issues.items() if f["reporter"] == "me"]
return [k for k, f in self.issues.items() if "AWPCA" in f["labels"]]
# --- маршрутизация ----------------------------------------------------------
def handle(self, h, method):
url = urlparse(h.path)
q = {k: v[0] for k, v in parse_qs(url.query).items()}
length = int(h.headers.get("Content-Length") or 0)
body = json.loads(h.rfile.read(length).decode("utf-8")) if length else {}
with self.lock:
self.requests.append((method, url.path))
if self.fail_next:
code, headers = self.fail_next.pop(0)
return self._send(h, code, {"errorMessages": ["имитация ошибки"]}, headers)
try:
code, payload = self.route(method, url.path, q, body)
except KeyError as e:
code, payload = 404, {"errorMessages": ["Issue Does Not Exist: %s" % e]}
self._send(h, code, payload)
def _send(self, h, code, payload, headers=None):
data = json.dumps(payload, ensure_ascii=False).encode("utf-8") if payload is not None else b""
h.send_response(code)
for k, v in (headers or {}).items():
h.send_header(k, v)
h.send_header("Content-Type", "application/json")
h.send_header("Content-Length", str(len(data)))
h.end_headers()
h.wfile.write(data)
def route(self, method, path, q, body):
p = path
if p == "/rest/api/2/myself":
return 200, {"name": "me", "displayName": "Me"}
if p == "/rest/api/2/search":
keys = self.search_keys(q.get("jql"))
start, maxr = int(q.get("startAt", 0)), int(q.get("maxResults", 50))
page = keys[start:start + maxr]
return 200, {"startAt": start, "maxResults": maxr, "total": len(keys), "issues": [self.issue_json(k) for k in page]}
if p == "/rest/api/2/issue" and method == "POST":
self.seq += 1
key = "%s-%d" % (self.project, self.seq)
f = body["fields"]
self.add(key, summary=f.get("summary"), description=f.get("description"),
priority=(f.get("priority") or {}).get("name", "Средний"), duedate=f.get("duedate"),
labels=list(f.get("labels") or []), issuetype=self._type_name(f.get("issuetype")),
reporter="me", created="2026-10-11T12:00:00.000+0300",
**{c: f.get(c) for c in CUSTOM})
return 201, {"id": "1", "key": key}
if p == "/rest/api/2/issueLinkType":
return 200, {"issueLinkTypes": LINK_TYPES}
if p == "/rest/api/2/issueLink" and method == "POST":
self.link_seq += 1
self.links.append({"id": self.link_seq, "type": body["type"]["name"],
"source": body["inwardIssue"]["key"], "target": body["outwardIssue"]["key"]})
return 201, {}
m = re.match(r"^/rest/api/2/issueLink/(\d+)$", p)
if m and method == "DELETE":
self.links = [l for l in self.links if str(l["id"]) != m.group(1)]
return 204, None
m = re.match(r"^/rest/api/2/issue/([^/]+)$", p)
if m:
key = m.group(1)
if method == "GET":
return 200, self.issue_json(key)
if method == "PUT":
f = self.issues[key]
for k, v in body.get("fields", {}).items():
if k in ("priority", "assignee"):
f[k] = (v or {}).get("name")
elif k == "labels":
f[k] = list(v)
else:
f[k] = v
f["updated"] = "2026-10-11T13:00:00.000+0300"
return 204, None
m = re.match(r"^/rest/api/2/issue/([^/]+)/comment$", p)
if m:
key = m.group(1)
self.issues[key]
lst = self.comments.setdefault(key, [])
if method == "GET":
return 200, {"comments": lst}
self.comment_seq += 1
lst.append({"id": str(self.comment_seq), "body": body["body"]})
return 201, lst[-1]
m = re.match(r"^/rest/api/2/issue/([^/]+)/comment/(\d+)$", p)
if m and method == "PUT":
for c in self.comments.get(m.group(1), []):
if c["id"] == m.group(2):
c["body"] = body["body"]
return 200, {}
m = re.match(r"^/rest/api/2/issue/([^/]+)/transitions$", p)
if m:
key = m.group(1)
trs = WORKFLOW.get(self.issues[key]["status"], [])
if method == "GET":
return 200, {"transitions": [{"id": str(i + 1), "name": n, "to": {"name": to}} for i, (n, to) in enumerate(trs)]}
tid = int(body["transition"]["id"]) - 1
self.issues[key]["status"] = trs[tid][1]
return 204, None
m = re.match(r"^/rest/agile/1\.0/board/(\d+)/sprint$", p)
if m:
states = (q.get("state") or "active,future").split(",")
return 200, {"values": [s for s in self.sprints if s["state"] in states], "isLast": True}
m = re.match(r"^/rest/agile/1\.0/sprint/(\d+)/issue$", p)
if m:
sp = next(s for s in self.sprints if s["id"] == int(m.group(1)))
if method == "GET":
keys = [k for k in self.search_keys(q.get("jql") or "key in ()") if self.issues[k]["sprint"] == sp["name"]]
return 200, {"issues": [{"key": k} for k in keys]}
for k in body["issues"]:
self.issues[k]["sprint"] = sp["name"]
return 204, None
if p == "/rest/agile/1.0/backlog/issue" and method == "POST":
for k in body["issues"]:
self.issues[k]["sprint"] = None
return 204, None
m = re.match(r"^/rest/agile/1\.0/issue/([^/]+)$", p)
if m:
name = self.issues[m.group(1)]["sprint"]
sp = next((s for s in self.sprints if s["name"] == name), None)
return 200, {"key": m.group(1), "fields": {"sprint": sp}}
return 404, {"errorMessages": ["нет маршрута %s %s" % (method, p)]}
def _type_name(self, ref):
ref = ref or {}
if ref.get("name"):
return ref["name"]
return {"10100": "Задача", "12300": "Фича"}.get(str(ref.get("id")), "Задача")
- Step 2: Тесты клиентов
tests/test_clients.py:
import unittest
from _common import agent, jira_cfg, no_sleep
from fake_jira import FakeJira
class ClientsTest(unittest.TestCase):
def setUp(self):
self.fake = FakeJira()
self.jira = agent.JiraClient(jira_cfg(self.fake))
self.jira.sleep = no_sleep
def tearDown(self):
self.fake.stop()
def test_get_and_counters(self):
self.fake.add("AURORA-1")
data = self.jira.get("/rest/api/2/issue/AURORA-1", fields="summary")
self.assertEqual(data["fields"]["summary"], "Тема AURORA-1")
self.assertEqual(self.jira.calls, 1)
self.assertIsNotNone(self.jira.last_call_at)
self.assertEqual(self.fake.requests[-1], ("GET", "/rest/api/2/issue/AURORA-1"))
def test_retry_on_429_with_retry_after(self):
self.fake.add("AURORA-1")
self.fake.fail_next.append((429, {"Retry-After": "0"}))
self.fake.fail_next.append((503, {}))
data = self.jira.get("/rest/api/2/issue/AURORA-1")
self.assertEqual(data["key"], "AURORA-1")
self.assertEqual(self.jira.rate_limited, 2)
self.assertEqual(self.jira.calls, 1)
def test_http_error_not_retried(self):
with self.assertRaises(agent.HttpError) as cm:
self.jira.get("/rest/api/2/issue/NOPE-1")
self.assertEqual(cm.exception.code, 404)
self.assertIn("Does Not Exist", cm.exception.text)
def test_gives_up_after_five(self):
self.fake.add("AURORA-1")
self.fake.fail_next.extend([(429, {"Retry-After": "0"})] * 5)
with self.assertRaises(agent.HttpError):
self.jira.get("/rest/api/2/issue/AURORA-1")
self.assertEqual(self.jira.rate_limited, 5)
def test_basic_auth_header_and_no_secret_in_repr(self):
self.assertTrue(self.jira.headers["Authorization"].startswith("Basic "))
self.assertNotIn("secret", repr(self.jira))
def test_load_config_requires_fields(self):
import json
import os
import tempfile
with tempfile.TemporaryDirectory() as d:
path = os.path.join(d, "agent.json")
with open(path, "w", encoding="utf-8") as f:
json.dump({"hub_url": "http://x", "hub_token": "t"}, f)
with self.assertRaises(SystemExit):
agent.load_config(path)
with open(path, "w", encoding="utf-8") as f:
json.dump(jira_cfg(self.fake), f)
cfg = agent.load_config(path)
self.assertEqual(cfg["poll_seconds"], 0)
def test_check_command(self):
import json
import os
import tempfile
with tempfile.TemporaryDirectory() as d:
path = os.path.join(d, "agent.json")
with open(path, "w", encoding="utf-8") as f:
json.dump(jira_cfg(self.fake), f)
# хаб недоступен (порт 1), Jira доступен — check печатает оба результата и возвращает 1
self.assertEqual(agent.main(["--config", path, "check"]), 1)
if __name__ == "__main__":
unittest.main()
- Step 3: Запустить — падает
Run: /usr/bin/python3 -m unittest tests.test_clients -v
Expected: FAIL — ModuleNotFoundError: No module named 'agent' (или AttributeError)
- Step 4: agent.py (скелет)
src/agent.example.json:
{
"hub_url": "http://192.168.0.106:8790",
"hub_token": "агентский токен из `jira-hub tokens`",
"jira": {
"base_url": "https://sdlc.go.rshbank.ru/jira",
"auth": {"type": "basic", "user": "логин", "password": "пароль", "token": ""}
},
"verify_ssl": true,
"ca_bundle": null,
"timeout": 60,
"poll_seconds": 5
}
src/agent.py:
#!/usr/bin/env python3
"""jira-agent — клиент моста Jira для виртуалки с доступом к Jira банка.
Опрашивает хаб (jira-hub) на Mac mini, выполняет команды «желаемый результат» в Jira по правилу
«проверь, потом сделай», ведёт журнал на диске и возвращает хабу состояние задач. Только стандартная библиотека,
Python 3.9. Спека: jira-hub/openspec/changes/bridge-core/specs/agent/spec.md.
python3 agent.py check # доступ к хабу и Jira, без записи
python3 agent.py run # основной цикл (по умолчанию)
python3 agent.py prune # почистить journal.jsonl от подтверждённых записей старше 30 дней
"""
import argparse
import base64
import json
import os
import signal
import ssl
import sys
import time
import urllib.error
import urllib.parse
import urllib.request
from datetime import datetime, timezone
from typing import Any, Dict, List, Optional
AGENT_VERSION = 1
HERE = os.path.dirname(os.path.abspath(__file__))
API = "/rest/api/2"
AGILE = "/rest/agile/1.0"
def now_iso() -> str:
return datetime.now(timezone.utc).replace(microsecond=0).isoformat()
def log(msg: str) -> None:
print("[%s] %s" % (datetime.now().strftime("%H:%M:%S"), msg), flush=True)
def load_config(path: str) -> dict:
try:
with open(path, encoding="utf-8") as f:
cfg = json.load(f)
except FileNotFoundError:
sys.exit("нет файла настроек %s — скопируйте agent.example.json в agent.json и заполните" % path)
for key in ("hub_url", "hub_token", "jira"):
if not cfg.get(key):
sys.exit("в %s нет обязательного поля %s" % (path, key))
if not cfg["jira"].get("base_url"):
sys.exit("в %s нет jira.base_url" % path)
cfg.setdefault("timeout", 60)
cfg.setdefault("poll_seconds", 5)
cfg.setdefault("verify_ssl", True)
return cfg
# ---------------------------------------------------------------------------
# HTTP
# ---------------------------------------------------------------------------
class HttpError(Exception):
def __init__(self, code: int, text: str, headers=None):
super().__init__("HTTP %d: %s" % (code, text[:300]))
self.code, self.text, self.headers = code, text, headers or {}
class Http:
def __init__(self, base: str, headers: Dict[str, str], timeout: float = 60, ctx=None):
self.base, self.headers, self.timeout, self.ctx = base.rstrip("/"), headers, timeout, ctx
def __repr__(self):
return "<%s %s>" % (self.__class__.__name__, self.base)
def call(self, method: str, path: str, body=None, params=None) -> Any:
url = self.base + path
if params:
url += "?" + urllib.parse.urlencode({k: v for k, v in params.items() if v is not None})
data = json.dumps(body, ensure_ascii=False).encode("utf-8") if body is not None else None
req = urllib.request.Request(url, data=data, method=method)
for k, v in self.headers.items():
req.add_header(k, v)
req.add_header("Accept", "application/json")
if data is not None:
req.add_header("Content-Type", "application/json; charset=utf-8")
try:
with urllib.request.urlopen(req, timeout=self.timeout, context=self.ctx) as resp:
raw = resp.read()
return json.loads(raw.decode("utf-8")) if raw else {}
except urllib.error.HTTPError as e:
raise HttpError(e.code, e.read().decode("utf-8", "replace"), dict(e.headers)) from None
except (urllib.error.URLError, ConnectionError, TimeoutError, OSError) as e:
raise ConnectionError("%s недоступен: %s" % (self.base, e)) from None
class JiraClient(Http):
"""Все вызовы Jira идут здесь: счётчики вызовов и 429/503, время последнего вызова, повтор по Retry-After."""
def __init__(self, cfg: dict):
j = cfg["jira"]
auth = j.get("auth") or {}
kind = auth.get("type") or ("bearer" if auth.get("token") else "basic")
if kind == "bearer":
if not auth.get("token"):
sys.exit("jira.auth.token пуст")
header = "Bearer " + auth["token"]
else:
if not (auth.get("user") and auth.get("password")):
sys.exit("jira.auth.user / jira.auth.password пусты")
header = "Basic " + base64.b64encode(("%s:%s" % (auth["user"], auth["password"])).encode()).decode()
ctx = None
if not cfg.get("verify_ssl", True):
ctx = ssl.create_default_context()
ctx.check_hostname = False
ctx.verify_mode = ssl.CERT_NONE
elif cfg.get("ca_bundle"):
ctx = ssl.create_default_context(cafile=cfg["ca_bundle"])
super().__init__(j["base_url"], {"Authorization": header, "X-Atlassian-Token": "no-check"}, cfg.get("timeout", 60), ctx)
self.calls = 0
self.rate_limited = 0
self.last_call_at = None # type: Optional[float] (time.monotonic)
self.last_call_iso = None # type: Optional[str]
self.sleep = time.sleep
def reset_counters(self) -> None:
self.calls, self.rate_limited = 0, 0
def call(self, method, path, body=None, params=None):
for attempt in range(1, 6):
try:
out = super().call(method, path, body, params)
self.calls += 1
self.last_call_at, self.last_call_iso = time.monotonic(), now_iso()
return out
except HttpError as e:
if e.code in (429, 503) and attempt < 5:
retry = e.headers.get("Retry-After")
wait = int(retry) if retry and retry.isdigit() else 60 * attempt
self.rate_limited += 1
log(" ! HTTP %d, жду %d с (попытка %d/5)" % (e.code, wait, attempt))
self.sleep(wait)
continue
if e.code in (429, 503):
self.rate_limited += 1
raise
def get(self, path, **params):
return self.call("GET", path, params=params or None)
def post(self, path, body):
return self.call("POST", path, body)
def put(self, path, body):
return self.call("PUT", path, body)
def delete(self, path):
return self.call("DELETE", path)
class HubClient(Http):
def __init__(self, cfg: dict):
super().__init__(cfg["hub_url"], {"Authorization": "Bearer " + cfg["hub_token"]}, cfg.get("timeout", 60))
def claim(self, agent_version, settings_version, last_jira_call_at) -> dict:
return self.call("POST", "/agent/claim", {"agent_version": agent_version, "settings_version": settings_version,
"last_jira_call_at": last_jira_call_at})
def settings(self) -> dict:
return self.call("GET", "/agent/settings")
def command(self, command_id) -> dict:
return self.call("GET", "/agent/commands/%d" % int(command_id))
def result(self, command_id, payload: dict) -> dict:
return self.call("POST", "/agent/commands/%d/result" % int(command_id), payload)
def state(self, payload: dict) -> dict:
return self.call("POST", "/agent/state", payload)
def known_keys(self) -> List[str]:
return self.call("GET", "/agent/known-keys")["keys"]
# ---------------------------------------------------------------------------
# CLI
# ---------------------------------------------------------------------------
def cmd_check(cfg: dict) -> int:
ok = True
try:
s = HubClient(cfg).settings()
log("хаб: доступен, настройки v%s, проект %s" % (s.get("settings_version"), s.get("project_key")))
except (ConnectionError, HttpError) as e:
log("хаб: НЕДОСТУПЕН — %s" % e)
ok = False
try:
me = JiraClient(cfg).get(API + "/myself")
log("Jira: доступен, пользователь %s" % me.get("name"))
except (ConnectionError, HttpError) as e:
log("Jira: НЕДОСТУПЕН — %s" % e)
ok = False
return 0 if ok else 1
def main(argv=None) -> int:
ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
ap.add_argument("--config", default=os.path.join(HERE, "agent.json"))
ap.add_argument("--journal", default=os.path.join(HERE, "journal.jsonl"))
ap.add_argument("cmd", nargs="?", default="run", choices=["run", "check", "prune"])
args = ap.parse_args(argv)
cfg = load_config(args.config)
if args.cmd == "check":
return cmd_check(cfg)
log("команда %s ещё не реализована" % args.cmd)
return 2
if __name__ == "__main__":
sys.exit(main())
- Step 5: Запустить — зелёный
Run: /usr/bin/python3 -m unittest tests.test_clients -v
Expected: 7 passed
- Step 6: Коммит
git add src/agent.py src/agent.example.json tests/__init__.py tests/_common.py tests/fake_jira.py tests/test_clients.py
git commit -m "feat(agent): скелет, клиенты Jira и хаба, фейковый Jira, команда check
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>"
Task 2: Журнал на диске и prune
Files:
- Modify:
src/agent.py(классJournalпослеHubClient;pruneвmain) - Test:
tests/test_journal.py
Interfaces:
-
Produces:
Journal(path):.append(record),.records() -> list,.started(command_id),.finished(command_id, status, result, error, jira_calls, rate_limited),.acked(command_id),.unacked_finished() -> list[dict](последняяfinishedна команду безackedпосле неё, в порядке журнала),.unfinished_started() -> list[int],.prune(days=30) -> int(сколько записей удалено). -
Step 1: Тест
tests/test_journal.py:
import json
import os
import tempfile
import unittest
from _common import agent
class JournalTest(unittest.TestCase):
def setUp(self):
self.dir = tempfile.TemporaryDirectory()
self.path = os.path.join(self.dir.name, "journal.jsonl")
self.j = agent.Journal(self.path)
def tearDown(self):
self.dir.cleanup()
def test_empty(self):
self.assertEqual(self.j.records(), [])
self.assertEqual(self.j.unacked_finished(), [])
self.assertEqual(self.j.unfinished_started(), [])
def test_lifecycle(self):
self.j.started(1)
self.assertEqual(self.j.unfinished_started(), [1])
self.j.finished(1, "done", {"key": "A-1"}, None, 2, 0)
self.assertEqual(self.j.unfinished_started(), [])
pending = self.j.unacked_finished()
self.assertEqual(len(pending), 1)
self.assertEqual(pending[0]["command_id"], 1)
self.assertEqual(pending[0]["result"], {"key": "A-1"})
self.assertEqual(pending[0]["jira_calls"], 2)
self.j.acked(1)
self.assertEqual(self.j.unacked_finished(), [])
self.j.started(2)
self.j.started(3)
self.j.finished(3, "failed", None, "HTTP 400", 1, 0)
self.assertEqual(self.j.unfinished_started(), [2])
self.assertEqual([r["command_id"] for r in self.j.unacked_finished()], [3])
def test_broken_line_is_skipped(self):
self.j.started(1)
with open(self.path, "a", encoding="utf-8") as f:
f.write("{не json\n")
self.j.finished(1, "done", None, None, 1, 0)
self.assertEqual(len(self.j.records()), 2)
def test_prune_keeps_unacked_and_recent(self):
self.j.started(1)
self.j.finished(1, "done", None, None, 1, 0)
self.j.append({"kind": "acked", "command_id": 1, "at": "2020-01-01T00:00:00+00:00"})
self.j.started(2)
self.j.finished(2, "done", None, None, 1, 0)
self.j.acked(2)
self.j.started(3)
removed = self.j.prune(days=30)
self.assertEqual(removed, 3)
ids = [(r["kind"], r["command_id"]) for r in self.j.records()]
self.assertEqual(ids, [("started", 2), ("finished", 2), ("acked", 2), ("started", 3)])
self.assertEqual(self.j.prune(days=30), 0)
def test_no_secrets_in_records(self):
self.j.finished(1, "done", {"key": "A-1"}, None, 1, 0)
with open(self.path, encoding="utf-8") as f:
text = f.read()
self.assertNotIn("password", text)
self.assertEqual(json.loads(text.splitlines()[0])["kind"], "finished")
if __name__ == "__main__":
unittest.main()
- Step 2: Запустить — падает
Run: /usr/bin/python3 -m unittest tests.test_journal -v
Expected: FAIL — AttributeError: module 'agent' has no attribute 'Journal'
- Step 3: Реализация
В src/agent.py после класса HubClient:
# ---------------------------------------------------------------------------
# Журнал: только дозапись; started → finished → acked по id команды
# ---------------------------------------------------------------------------
class Journal:
def __init__(self, path: str):
self.path = path
def append(self, record: dict) -> None:
record = dict(record)
record.setdefault("at", now_iso())
with open(self.path, "a", encoding="utf-8") as f:
f.write(json.dumps(record, ensure_ascii=False) + "\n")
def records(self) -> List[dict]:
if not os.path.exists(self.path):
return []
out = []
with open(self.path, encoding="utf-8") as f:
for n, line in enumerate(f, 1):
line = line.strip()
if not line:
continue
try:
out.append(json.loads(line))
except ValueError:
log("журнал: строка %d не читается, пропускаю" % n)
return out
def started(self, command_id) -> None:
self.append({"kind": "started", "command_id": int(command_id)})
def finished(self, command_id, status, result, error, jira_calls, rate_limited) -> None:
self.append({"kind": "finished", "command_id": int(command_id), "status": status, "result": result,
"error": error, "jira_calls": int(jira_calls or 0), "rate_limited": int(rate_limited or 0)})
def acked(self, command_id) -> None:
self.append({"kind": "acked", "command_id": int(command_id)})
def unacked_finished(self) -> List[dict]:
last, acked = {}, set()
for r in self.records():
if r.get("kind") == "finished":
last[r["command_id"]] = r
acked.discard(r["command_id"])
elif r.get("kind") == "acked":
acked.add(r["command_id"])
return [r for cid, r in last.items() if cid not in acked]
def unfinished_started(self) -> List[int]:
started, finished = [], set()
for r in self.records():
if r.get("kind") == "started":
started.append(r["command_id"])
elif r.get("kind") in ("finished", "acked"):
finished.add(r["command_id"])
return [cid for cid in started if cid not in finished]
def prune(self, days: int = 30) -> int:
"""Удалить записи команд, подтверждённых (acked) раньше, чем days дней назад. Неподтверждённые — остаются."""
records = self.records()
cutoff = datetime.now(timezone.utc).timestamp() - days * 86400
old = set()
for r in records:
if r.get("kind") == "acked":
try:
ts = datetime.fromisoformat(r.get("at", "")).timestamp()
except ValueError:
continue
if ts < cutoff:
old.add(r["command_id"])
keep = [r for r in records if r.get("command_id") not in old]
tmp = self.path + ".tmp"
with open(tmp, "w", encoding="utf-8") as f:
for r in keep:
f.write(json.dumps(r, ensure_ascii=False) + "\n")
os.replace(tmp, self.path)
return len(records) - len(keep)
В main заменить заглушку:
if args.cmd == "prune":
removed = Journal(args.journal).prune(30)
log("журнал: удалено записей %d" % removed)
return 0
log("команда %s ещё не реализована" % args.cmd)
return 2
- Step 4: Запустить — зелёный
Run: /usr/bin/python3 -m unittest tests.test_journal -v
Expected: 5 passed
- Step 5: Коммит
git add src/agent.py tests/test_journal.py
git commit -m "feat(agent): журнал на диске, восстановление по нему, prune
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>"
Task 3: Исполнитель — ensure_issue, ensure_fields, ensure_comment
Files:
- Modify:
src/agent.py(классExecutorпослеJournal) - Test:
tests/test_executor_fields.py
Interfaces:
-
Produces:
Executor(jira, settings, hub=None):.run(cmd: dict) -> {"status", "result", "error", "jira_calls", "rate_limited"},.sleep(подменяемый),.gap(cmd_type),.fid(logical),.priority_value(name),.issue_type_ref(kind),.epic_fields(epic_key), методыdo_ensure_issue(p, cmd),do_ensure_fields(p, cmd),do_ensure_comment(p, cmd). -
Правило сравнения дат: по первым 10 символам (
YYYY-MM-DD), потому что кастомные даты Jira приходят со временем. -
Step 1: Тесты
tests/test_executor_fields.py:
import unittest
from _common import SETTINGS, agent, jira_cfg, no_sleep
from fake_jira import FakeJira
class ExecutorFieldsTest(unittest.TestCase):
def setUp(self):
self.fake = FakeJira()
self.jira = agent.JiraClient(jira_cfg(self.fake))
self.jira.sleep = no_sleep
self.ex = agent.Executor(self.jira, dict(SETTINGS))
self.ex.sleep = no_sleep
def tearDown(self):
self.fake.stop()
def run_cmd(self, ctype, **params):
return self.ex.run({"id": 1, "type": ctype, "params": params})
def test_ensure_issue_creates_then_finds(self):
out = self.run_cmd("ensure_issue", temp_key="new:1", issue_type="task", summary="Новая задача", description="h3. Контекст",
priority="Высокий", duedate="2026-10-23", planned_start="2026-10-20", labels=["AWPCA", "kind-dev"],
epic_key="AURORA-1350", assignee="ivanov")
self.assertEqual(out["status"], "done")
key = out["result"]["key"]
self.assertTrue(out["result"]["created"])
f = self.fake.issues[key]
self.assertEqual(f["priority"], "Высокий")
self.assertEqual(f["customfield_10801"], "2026-10-20")
self.assertEqual(f["customfield_15302"], "AURORA-1350")
self.assertEqual(f["labels"], ["AWPCA", "kind-dev"])
self.assertEqual(out["jira_calls"], 2)
again = self.run_cmd("ensure_issue", temp_key="new:1", issue_type="task", summary="Новая задача")
self.assertEqual(again["status"], "done")
self.assertEqual(again["result"], {"key": key, "existed": True})
self.assertEqual(len(self.fake.issues), 1)
self.assertEqual(again["jira_calls"], 1)
def test_ensure_issue_epic_sets_epic_name(self):
out = self.run_cmd("ensure_issue", temp_key="new:2", issue_type="epic", summary="Приёмка")
f = self.fake.issues[out["result"]["key"]]
self.assertEqual(f["issuetype"], "Фича")
self.assertEqual(f["customfield_15304"], "Приёмка")
def test_ensure_issue_priority_mapped(self):
out = self.run_cmd("ensure_issue", temp_key="new:3", issue_type="task", summary="x", priority="Наивысший")
self.assertEqual(self.fake.issues[out["result"]["key"]]["priority"], "Критичный")
def test_ensure_fields_writes_only_diff(self):
self.fake.add("AURORA-1", duedate="2026-10-20", labels=["AWPCA", "fact-open"], priority="Средний")
out = self.run_cmd("ensure_fields", key="AURORA-1", duedate="2026-10-23", priority="Средний",
labels_add=["kind-dev", "AWPCA"], labels_remove=["fact-open"], assignee="ivanov",
planned_end="2026-10-23", epic_key="AURORA-1350")
self.assertEqual(out["status"], "done")
self.assertEqual(out["result"]["changed"], ["duedate", "planned_end", "epic_key", "assignee", "labels"])
f = self.fake.issues["AURORA-1"]
self.assertEqual(f["labels"], ["AWPCA", "kind-dev"])
self.assertEqual(f["assignee"], "ivanov")
self.assertEqual(f["customfield_10802"], "2026-10-23")
self.assertEqual(len(self.fake.writes()), 1)
again = self.run_cmd("ensure_fields", key="AURORA-1", duedate="2026-10-23", labels_add=["kind-dev"], assignee="ivanov")
self.assertEqual(again["result"], {"changed": []})
self.assertEqual(len(self.fake.writes()), 1)
def test_fields_description_exact_compare(self):
text = "h3. Контекст\n* пункт *жирный*\n\nh3. Что сделать"
self.fake.add("AURORA-1", description=text)
out = self.run_cmd("ensure_fields", key="AURORA-1", description=text)
self.assertEqual(out["result"]["changed"], [])
out = self.run_cmd("ensure_fields", key="AURORA-1", description=text + "\n")
self.assertEqual(out["result"]["changed"], ["description"])
def test_fields_date_with_time_from_jira(self):
self.fake.add("AURORA-1", customfield_10802="2026-10-23T00:00:00.000+0300")
out = self.run_cmd("ensure_fields", key="AURORA-1", planned_end="2026-10-23")
self.assertEqual(out["result"]["changed"], [])
def test_fields_clear_assignee_and_epic(self):
self.fake.add("AURORA-1", assignee="ivanov", customfield_15302="AURORA-1350")
out = self.run_cmd("ensure_fields", key="AURORA-1", assignee=None, epic_key=None)
self.assertEqual(out["result"]["changed"], ["epic_key", "assignee"])
self.assertIsNone(self.fake.issues["AURORA-1"]["assignee"])
self.assertIsNone(self.fake.issues["AURORA-1"]["customfield_15302"])
def test_fields_missing_issue_is_failed(self):
out = self.run_cmd("ensure_fields", key="AURORA-404", duedate="2026-10-23")
self.assertEqual(out["status"], "failed")
self.assertIn("HTTP 404", out["error"])
def test_ensure_comment_upsert(self):
self.fake.add("AURORA-1")
out = self.run_cmd("ensure_comment", key="AURORA-1", marker="Актуализация 10.10.2026", body="Сделано: список")
self.assertEqual(out["status"], "done")
self.assertTrue(out["result"]["created"])
self.assertEqual(self.fake.comments["AURORA-1"][0]["body"], "*Актуализация 10.10.2026.* Сделано: список")
same = self.run_cmd("ensure_comment", key="AURORA-1", marker="Актуализация 10.10.2026", body="Сделано: список")
self.assertEqual(same["result"]["changed"], False)
upd = self.run_cmd("ensure_comment", key="AURORA-1", marker="Актуализация 10.10.2026", body="Сделано: всё")
self.assertEqual(upd["result"]["changed"], True)
self.assertEqual(len(self.fake.comments["AURORA-1"]), 1)
self.assertEqual(self.fake.comments["AURORA-1"][0]["body"], "*Актуализация 10.10.2026.* Сделано: всё")
def test_unknown_type(self):
self.assertEqual(self.run_cmd("ensure_magic", key="x")["status"], "failed")
if __name__ == "__main__":
unittest.main()
- Step 2: Запустить — падает
Run: /usr/bin/python3 -m unittest tests.test_executor_fields -v
Expected: FAIL — AttributeError: module 'agent' has no attribute 'Executor'
- Step 3: Реализация
В src/agent.py после класса Journal:
# ---------------------------------------------------------------------------
# Исполнитель: команда — желаемый результат; прочитать → сравнить → записать разницу
# ---------------------------------------------------------------------------
def _date(value) -> Optional[str]:
"""Даты Jira (duedate — YYYY-MM-DD, кастомные — с временем) сравниваем по дню."""
return (value or "")[:10] or None
class Executor:
def __init__(self, jira: JiraClient, settings: dict, hub: Optional[HubClient] = None):
self.jira, self.s, self.hub = jira, settings, hub
self.sleep = time.sleep
self._symmetric = None
self._sprints = None
def run(self, cmd: dict) -> dict:
self.jira.reset_counters()
self._sprints = None
handler = getattr(self, "do_" + str(cmd.get("type")), None)
if handler is None:
return self._out("failed", error="неизвестный тип команды %s" % cmd.get("type"))
try:
return handler(cmd.get("params") or {}, cmd)
except HttpError as e:
return self._out("failed", error="HTTP %d %s" % (e.code, e.text[:1500]))
except ConnectionError as e:
return self._out("failed", error=str(e))
def _out(self, status, result=None, error=None) -> dict:
return {"status": status, "result": result, "error": error,
"jira_calls": self.jira.calls, "rate_limited": self.jira.rate_limited}
def gap(self, cmd_type: str) -> None:
"""Зазор между записями внутри одной команды — по политике хаба для этого типа."""
seconds = int((self.s.get("pacing") or {}).get(cmd_type, 0) or 0)
if seconds > 0:
log(" пауза %d с" % seconds)
self.sleep(seconds)
def fid(self, logical: str) -> str:
return (self.s.get("field_ids") or {}).get(logical, logical)
def has_field(self, logical: str) -> bool:
return logical in ("summary", "description", "priority", "duedate", "labels") or bool((self.s.get("field_ids") or {}).get(logical))
def priority_value(self, name: str) -> dict:
mapped = (self.s.get("priority_map") or {}).get(name, name)
return {self.s.get("priority_value_key") or "name": mapped}
def issue_type_ref(self, kind: str) -> dict:
it = (self.s.get("issue_types") or {}).get(kind) or {}
return {"id": str(it["id"])} if it.get("id") else {"name": it.get("name")}
def epic_fields(self, epic_key) -> dict:
ep = self.s.get("epic_link") or {}
if epic_key and ep.get("mode", "field") == "field" and ep.get("field_id"):
return {ep["field_id"]: epic_key}
return {}
# --- ensure_issue ----------------------------------------------------------
def do_ensure_issue(self, p: dict, cmd: dict) -> dict:
jql = "project = %s AND reporter = currentUser() AND created >= -7d" % self.s["project_key"]
found = self.jira.get(API + "/search", jql=jql, fields="summary", maxResults=200)
for it in found.get("issues", []):
if (it.get("fields", {}).get("summary") or "") == p["summary"]:
return self._out("done", {"key": it["key"], "existed": True})
fields = {"project": {"key": self.s["project_key"]}, "issuetype": self.issue_type_ref(p["issue_type"]),
"summary": p["summary"]}
if p.get("description") is not None:
fields[self.fid("description")] = p["description"]
if p.get("priority"):
fields[self.fid("priority")] = self.priority_value(p["priority"])
if p.get("duedate"):
fields[self.fid("duedate")] = p["duedate"]
for logical in ("planned_start", "planned_end"):
if p.get(logical) and self.has_field(logical):
fields[self.fid(logical)] = p[logical]
if p.get("labels") is not None:
fields[self.fid("labels")] = list(p["labels"])
if p.get("assignee"):
fields["assignee"] = {"name": p["assignee"]}
fields.update(self.epic_fields(p.get("epic_key")))
if p["issue_type"] == "epic" and self.s.get("epic_name_field_id"):
fields[self.s["epic_name_field_id"]] = p["summary"]
created = self.jira.post(API + "/issue", {"fields": fields})
return self._out("done", {"key": created["key"], "created": True})
# --- ensure_fields ---------------------------------------------------------
def do_ensure_fields(self, p: dict, cmd: dict) -> dict:
key = p["key"]
cur = self.jira.get(API + "/issue/%s" % key)["fields"]
fields, changed = {}, []
if "summary" in p and (cur.get("summary") or "") != (p["summary"] or ""):
fields["summary"] = p["summary"]
changed.append("summary")
if "description" in p and (cur.get(self.fid("description")) or "") != (p["description"] or ""):
fields[self.fid("description")] = p["description"]
changed.append("description")
if "priority" in p:
want = self.priority_value(p["priority"])
vkey = list(want)[0]
if ((cur.get(self.fid("priority")) or {}).get(vkey)) != want[vkey]:
fields[self.fid("priority")] = want
changed.append("priority")
for logical in ("duedate", "planned_start", "planned_end"):
if logical in p and self.has_field(logical) and _date(cur.get(self.fid(logical))) != _date(p[logical]):
fields[self.fid(logical)] = p[logical]
changed.append(logical)
if "epic_key" in p:
ep = self.s.get("epic_link") or {}
if ep.get("field_id") and (cur.get(ep["field_id"]) or None) != (p["epic_key"] or None):
fields[ep["field_id"]] = p["epic_key"]
changed.append("epic_key")
if "assignee" in p and ((cur.get("assignee") or {}).get("name") or None) != (p["assignee"] or None):
fields["assignee"] = {"name": p["assignee"]} if p["assignee"] else None
changed.append("assignee")
if "labels_add" in p or "labels_remove" in p:
labels = list(cur.get(self.fid("labels")) or [])
remove = set(p.get("labels_remove") or [])
new = [l for l in labels if l not in remove] + [l for l in (p.get("labels_add") or []) if l not in labels]
if new != labels:
fields[self.fid("labels")] = new
changed.append("labels")
if not fields:
return self._out("done", {"changed": []})
self.jira.put(API + "/issue/%s" % key, {"fields": fields})
return self._out("done", {"changed": changed})
# --- ensure_comment --------------------------------------------------------
def do_ensure_comment(self, p: dict, cmd: dict) -> dict:
key, marker, body = p["key"], p["marker"], p["body"]
if marker not in body[:80]:
body = "*%s.* %s" % (marker, body)
existing = None
for c in self.jira.get(API + "/issue/%s/comment" % key).get("comments", []):
if marker in (c.get("body") or "")[:80]:
existing = c
break
if existing and (existing.get("body") or "") == body:
return self._out("done", {"comment_id": existing["id"], "changed": False})
if existing:
self.jira.put(API + "/issue/%s/comment/%s" % (key, existing["id"]), {"body": body})
return self._out("done", {"comment_id": existing["id"], "changed": True})
c = self.jira.post(API + "/issue/%s/comment" % key, {"body": body})
return self._out("done", {"comment_id": c.get("id"), "created": True})
- Step 4: Запустить — зелёный
Run: /usr/bin/python3 -m unittest tests.test_executor_fields -v
Expected: 10 passed
- Step 5: Коммит
git add src/agent.py tests/test_executor_fields.py
git commit -m "feat(agent): исполнитель — ensure_issue, ensure_fields, ensure_comment
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>"
Task 4: Исполнитель — ensure_status, ensure_link, ensure_sprint
Files:
- Modify:
src/agent.py(методы вExecutor) - Test:
tests/test_executor_links.py
Interfaces:
-
Produces:
Executor.do_ensure_status,.do_ensure_link,.do_ensure_sprint,.symmetric_types() -> set,.sprints() -> dict[name, sprint](активные и будущие доскиboard_id;{}без доски). -
Step 1: Тесты
tests/test_executor_links.py:
import unittest
from _common import SETTINGS, agent, jira_cfg, no_sleep
from fake_jira import FakeJira
class ExecutorLinksTest(unittest.TestCase):
def setUp(self):
self.fake = FakeJira()
self.jira = agent.JiraClient(jira_cfg(self.fake))
self.jira.sleep = no_sleep
self.ex = agent.Executor(self.jira, dict(SETTINGS))
self.gaps = []
self.ex.sleep = self.gaps.append
self.fake.add("AURORA-1")
self.fake.add("AURORA-2")
def tearDown(self):
self.fake.stop()
def run_cmd(self, ctype, **params):
return self.ex.run({"id": 1, "type": ctype, "params": params})
# --- статус ---
def test_status_path(self):
out = self.run_cmd("ensure_status", key="AURORA-1", target=["Разработка", "Готово"])
self.assertEqual(out["status"], "done")
self.assertEqual(out["result"], {"from": "Ожидание разработки", "to": "Готово", "steps": 2})
self.assertEqual(self.fake.issues["AURORA-1"]["status"], "Готово")
again = self.run_cmd("ensure_status", key="AURORA-1", target=["Разработка", "Готово"])
self.assertEqual(again["result"]["steps"], 0)
self.assertEqual(again["jira_calls"], 1)
def test_status_resumes_from_middle(self):
self.fake.issues["AURORA-1"]["status"] = "Разработка"
out = self.run_cmd("ensure_status", key="AURORA-1", target=["Разработка", "Готово к тестированию"])
self.assertEqual(out["result"]["steps"], 1)
self.assertEqual(self.fake.issues["AURORA-1"]["status"], "Готово к тестированию")
def test_status_single_and_by_transition_name(self):
out = self.run_cmd("ensure_status", key="AURORA-1", target="Взять в разработку")
self.assertEqual(out["result"]["to"], "Взять в разработку") # имя перехода принято как шаг
self.assertEqual(self.fake.issues["AURORA-1"]["status"], "Разработка")
def test_status_no_transition_is_failed(self):
out = self.run_cmd("ensure_status", key="AURORA-1", target="Тестирование")
self.assertEqual(out["status"], "failed")
self.assertIn("нет перехода из «Ожидание разработки» в «Тестирование»", out["error"])
self.assertIn("Взять в разработку→Разработка", out["error"])
self.assertEqual(out["result"], {"from": "Ожидание разработки", "reached": "Ожидание разработки"})
def test_status_gap_between_steps(self):
self.ex.s["pacing"] = {"ensure_status": 7}
self.run_cmd("ensure_status", key="AURORA-1", target=["Разработка", "Готово"])
self.assertEqual(self.gaps, [7])
# --- связи ---
def test_link_create_and_idempotent(self):
out = self.run_cmd("ensure_link", source="AURORA-1", target="AURORA-2", link_type="блокирует начало")
self.assertEqual(out["result"], {"created": True, "removed": []})
link = self.fake.links[0]
self.assertEqual((link["source"], link["target"], link["type"]), ("AURORA-1", "AURORA-2", "ASUPR_Блокер_Начала"))
again = self.run_cmd("ensure_link", source="AURORA-1", target="AURORA-2", link_type="блокирует начало")
self.assertEqual(again["result"], {"existed": True, "removed": []})
self.assertEqual(len(self.fake.links), 1)
def test_link_symmetric_any_direction(self):
self.fake.links.append({"id": 1, "type": "ASUPR_Связанные", "source": "AURORA-2", "target": "AURORA-1"})
out = self.run_cmd("ensure_link", source="AURORA-1", target="AURORA-2", link_type="связанные")
self.assertEqual(out["result"], {"existed": True, "removed": []})
def test_link_conflict_without_replace(self):
self.fake.links.append({"id": 1, "type": "ASUPR_Блокер_Начала", "source": "AURORA-2", "target": "AURORA-1"})
out = self.run_cmd("ensure_link", source="AURORA-1", target="AURORA-2", link_type="блокирует начало")
self.assertEqual(out["status"], "failed")
self.assertIn("обратное направление", out["error"])
self.assertEqual(len(self.fake.links), 1)
def test_link_replace_reverse_and_other_type(self):
self.ex.s["pacing"] = {"ensure_link": 3}
self.fake.links.append({"id": 1, "type": "ASUPR_Блокер_Начала", "source": "AURORA-2", "target": "AURORA-1"})
self.fake.links.append({"id": 2, "type": "ASUPR_Связанные", "source": "AURORA-1", "target": "AURORA-2"})
out = self.run_cmd("ensure_link", source="AURORA-1", target="AURORA-2", link_type="блокирует начало", replace=True)
self.assertEqual(out["status"], "done")
self.assertEqual(sorted(out["result"]["removed"]), ["ASUPR_Блокер_Начала", "ASUPR_Связанные"])
self.assertTrue(out["result"]["created"])
self.assertEqual([(l["source"], l["target"], l["type"]) for l in self.fake.links],
[("AURORA-1", "AURORA-2", "ASUPR_Блокер_Начала")])
self.assertEqual(self.gaps, [3, 3]) # между двумя DELETE и перед POST
self.assertEqual([r[0] for r in self.fake.writes()], ["DELETE", "DELETE", "POST"])
def test_link_unmanaged_type_ignored(self):
self.fake.links.append({"id": 1, "type": "Duplicate", "source": "AURORA-1", "target": "AURORA-2"})
out = self.run_cmd("ensure_link", source="AURORA-1", target="AURORA-2", link_type="связанные")
self.assertEqual(out["result"], {"created": True, "removed": []})
self.assertEqual(len(self.fake.links), 2)
# --- спринты ---
def test_sprint_move_and_idempotent(self):
out = self.run_cmd("ensure_sprint", key="AURORA-1", sprint="Спринт 2")
self.assertEqual(out["result"], {"sprint": "Спринт 2", "from": None})
self.assertEqual(self.fake.issues["AURORA-1"]["sprint"], "Спринт 2")
again = self.run_cmd("ensure_sprint", key="AURORA-1", sprint="Спринт 2")
self.assertEqual(again["result"], {"sprint": "Спринт 2", "changed": False})
self.assertEqual(again["jira_calls"], 1)
def test_sprint_backlog(self):
self.fake.issues["AURORA-1"]["sprint"] = "Спринт 1"
out = self.run_cmd("ensure_sprint", key="AURORA-1", sprint="backlog")
self.assertEqual(out["result"], {"sprint": None, "from": "Спринт 1"})
self.assertIsNone(self.fake.issues["AURORA-1"]["sprint"])
self.assertEqual(self.run_cmd("ensure_sprint", key="AURORA-1", sprint="backlog")["result"], {"sprint": None, "changed": False})
def test_sprint_unknown_name(self):
out = self.run_cmd("ensure_sprint", key="AURORA-1", sprint="Спринт 9")
self.assertEqual(out["status"], "failed")
self.assertIn("Спринт 9", out["error"])
self.assertIn("Спринт 2", out["error"])
def test_sprint_without_board(self):
self.ex.s["board_id"] = None
out = self.run_cmd("ensure_sprint", key="AURORA-1", sprint="Спринт 2")
self.assertEqual(out["status"], "failed")
self.assertIn("board_id", out["error"])
if __name__ == "__main__":
unittest.main()
- Step 2: Запустить — падает
Run: /usr/bin/python3 -m unittest tests.test_executor_links -v
Expected: FAIL — AttributeError: 'Executor' object has no attribute 'do_ensure_status' (команды failed с «неизвестный тип»)
- Step 3: Реализация
В класс Executor после do_ensure_comment:
# --- ensure_status ---------------------------------------------------------
def do_ensure_status(self, p: dict, cmd: dict) -> dict:
key, target = p["key"], p["target"]
path = list(target) if isinstance(target, list) else [target]
final = path[-1]
cur = self.jira.get(API + "/issue/%s" % key, fields="status")["fields"]["status"]["name"]
start = cur
if cur == final:
return self._out("done", {"from": start, "to": final, "steps": 0})
if cur in path:
path = path[path.index(cur) + 1:]
steps = 0
for step in path:
if cur == step:
continue
trs = self.jira.get(API + "/issue/%s/transitions" % key).get("transitions", [])
match = [t for t in trs if (t.get("to") or {}).get("name") == step or t.get("name") == step]
if not match:
avail = ", ".join("%s→%s" % (t.get("name"), (t.get("to") or {}).get("name")) for t in trs)
return self._out("failed", {"from": start, "reached": cur},
"нет перехода из «%s» в «%s»; доступны: %s" % (cur, step, avail or "—"))
if steps:
self.gap("ensure_status")
self.jira.post(API + "/issue/%s/transitions" % key, {"transition": {"id": match[0]["id"]}})
cur = step
steps += 1
return self._out("done", {"from": start, "to": cur, "steps": steps})
# --- ensure_link -----------------------------------------------------------
def symmetric_types(self) -> set:
"""Типы, у которых inward == outward («Связанные»): направление не важно."""
if self._symmetric is None:
try:
types = self.jira.get(API + "/issueLinkType").get("issueLinkTypes", [])
self._symmetric = {t["name"] for t in types if t.get("inward") == t.get("outward")}
except HttpError:
self._symmetric = set()
return self._symmetric
def do_ensure_link(self, p: dict, cmd: dict) -> dict:
src, dst = p["source"], p["target"]
types = self.s.get("link_types") or {}
lt = types.get(p["link_type"], p["link_type"])
managed = set(types.values()) | {lt}
symmetric = self.symmetric_types()
raw = self.jira.get(API + "/issue/%s" % src, fields="issuelinks")["fields"].get("issuelinks") or []
links = [] # (тип, другая задача, направление, id)
for l in raw:
t = (l.get("type") or {}).get("name")
if l.get("outwardIssue"):
links.append((t, l["outwardIssue"]["key"], "out", l.get("id")))
if l.get("inwardIssue"):
links.append((t, l["inwardIssue"]["key"], "in", l.get("id")))
wanted = [x for x in links if x[1] == dst and x[0] == lt and (x[2] == "out" or lt in symmetric)]
conflicts = [x for x in links if x[1] == dst and x[0] in managed and x not in wanted]
if conflicts and not p.get("replace"):
desc = ", ".join("«%s» (%s)" % (t, "обратное направление" if t == lt else "другой тип") for t, _, _, _ in conflicts)
return self._out("failed", error="между %s и %s уже есть: %s; нужен replace" % (src, dst, desc))
removed = []
for t, other, direction, lid in conflicts:
if not lid:
return self._out("failed", {"removed": removed}, "у связи «%s» с %s нет id — снять нельзя" % (t, other))
if removed:
self.gap("ensure_link")
self.jira.delete(API + "/issueLink/%s" % lid)
removed.append(t)
if wanted:
return self._out("done", {"existed": True, "removed": removed})
if removed:
self.gap("ensure_link")
# inwardIssue — источник (получает outward-описание «Блокирует начало…»), outwardIssue — цель
self.jira.post(API + "/issueLink", {"type": {"name": lt}, "inwardIssue": {"key": src}, "outwardIssue": {"key": dst}})
return self._out("done", {"created": True, "removed": removed})
# --- ensure_sprint ---------------------------------------------------------
def sprints(self) -> Dict[str, dict]:
"""Имя → спринт (active и future) доски board_id; без доски — пусто."""
if self._sprints is None:
out, start = {}, 0
board = self.s.get("board_id")
while board:
page = self.jira.get(AGILE + "/board/%s/sprint" % board, state="active,future", startAt=start, maxResults=50)
for sp in page.get("values", []):
out[sp["name"]] = sp
if page.get("isLast", True) or not page.get("values"):
break
start += len(page["values"])
self._sprints = out
return self._sprints
def do_ensure_sprint(self, p: dict, cmd: dict) -> dict:
key, want = p["key"], p["sprint"]
if not self.s.get("board_id"):
return self._out("failed", error="board_id не задан в настройках хаба — спринты недоступны")
cur = ((self.jira.get(AGILE + "/issue/%s" % key, fields="sprint")["fields"].get("sprint") or {}).get("name"))
if want == "backlog":
if cur is None:
return self._out("done", {"sprint": None, "changed": False})
self.jira.post(AGILE + "/backlog/issue", {"issues": [key]})
return self._out("done", {"sprint": None, "from": cur})
if cur == want:
return self._out("done", {"sprint": want, "changed": False})
sp = self.sprints().get(want)
if not sp:
return self._out("failed", error="нет спринта «%s» среди активных и будущих доски %s: %s"
% (want, self.s["board_id"], ", ".join(sorted(self.sprints())) or "—"))
self.jira.post(AGILE + "/sprint/%s/issue" % sp["id"], {"issues": [key]})
return self._out("done", {"sprint": want, "from": cur})
- Step 4: Запустить — зелёный
Run: /usr/bin/python3 -m unittest tests.test_executor_links -v
Expected: 14 passed
- Step 5: Коммит
git add src/agent.py tests/test_executor_links.py
git commit -m "feat(agent): исполнитель — переходы статусов, связи с заменой, спринты
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>"
Task 5: Чтение состояния (refresh)
Files:
- Modify:
src/agent.py(методы вExecutor) - Create:
tests/fake_hub.py - Test:
tests/test_refresh.py
Interfaces:
-
Produces:
Executor.state_fields() -> str,.search_all(jql) -> list[raw],.to_state(raw) -> dict(ключи зеркала хаба),.links_from(raw) -> list[{"source_key","target_key","link_type"}],.sprint_names(keys) -> dict,.do_refresh(p, cmd)(шлётhub.state(...)страницами по 100, возвращает{"issues", "neighbors", "pages"}). -
Produces (тесты):
fake_hub.FakeHub(settings):.base_url,.token = "agent-token",.enqueue(type, params, pause=0) -> id,.commands,.results,.states,.known,.claims,.fail_results(сколько раз ответить 503 на отчёт),.reject_results({id: код}),.stop(). -
Step 1: Фейковый хаб
tests/fake_hub.py:
"""Фейковый хаб для тестов агента: маршруты /agent/* в памяти."""
import json
import re
import threading
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
class FakeHub:
token = "agent-token"
def __init__(self, settings):
self.settings = dict(settings)
self.queue = []
self.commands = {}
self.results = {}
self.states = []
self.known = []
self.claims = 0
self.fail_results = 0
self.reject_results = {}
self.in_flight = None
self.lock = threading.Lock()
hub = self
class Handler(BaseHTTPRequestHandler):
def log_message(self, *a):
pass
def do_GET(self):
hub.handle(self, "GET")
def do_POST(self):
hub.handle(self, "POST")
self.server = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
threading.Thread(target=self.server.serve_forever, daemon=True).start()
@property
def base_url(self):
return "http://127.0.0.1:%d" % self.server.server_port
def stop(self):
self.server.shutdown()
def enqueue(self, ctype, params, pause=0):
cid = len(self.commands) + 1
self.commands[cid] = {"id": cid, "type": ctype, "params": params, "pause_seconds": pause, "status": "queued"}
self.queue.append(cid)
return cid
def handle(self, h, method):
length = int(h.headers.get("Content-Length") or 0)
body = json.loads(h.rfile.read(length).decode("utf-8")) if length else {}
if h.headers.get("Authorization") != "Bearer " + self.token:
return self._send(h, 401, {"error": "unauthorized"})
with self.lock:
code, payload = self.route(method, h.path.split("?")[0], body)
self._send(h, code, payload)
def _send(self, h, code, payload):
data = json.dumps(payload, ensure_ascii=False).encode("utf-8")
h.send_response(code)
h.send_header("Content-Type", "application/json")
h.send_header("Content-Length", str(len(data)))
h.end_headers()
h.wfile.write(data)
def route(self, method, path, body):
if path == "/agent/claim":
self.claims += 1
if self.in_flight and self.commands[self.in_flight]["status"] == "claimed":
return 200, {"command": None, "settings_version": self.settings["settings_version"]}
if not self.queue:
return 200, {"command": None, "settings_version": self.settings["settings_version"]}
cid = self.queue.pop(0)
self.commands[cid]["status"] = "claimed"
self.in_flight = cid
c = self.commands[cid]
return 200, {"command": {"id": cid, "type": c["type"], "params": c["params"], "pause_seconds": c["pause_seconds"]},
"settings_version": self.settings["settings_version"]}
if path == "/agent/settings":
return 200, self.settings
if path == "/agent/known-keys":
return 200, {"keys": list(self.known)}
if path == "/agent/state":
self.states.append(body)
return 200, {"issues": len(body.get("issues") or []), "links": len(body.get("links") or [])}
m = re.match(r"^/agent/commands/(\d+)$", path)
if m:
c = self.commands.get(int(m.group(1)))
return (200, c) if c else (404, {"error": "not_found"})
m = re.match(r"^/agent/commands/(\d+)/result$", path)
if m:
cid = int(m.group(1))
if self.fail_results > 0:
self.fail_results -= 1
return 503, {"error": "busy"}
if cid in self.reject_results:
return self.reject_results[cid], {"error": "rejected"}
if cid not in self.commands:
return 404, {"error": "not_found"}
self.results[cid] = body
self.commands[cid]["status"] = body.get("status")
if self.in_flight == cid:
self.in_flight = None
return 200, self.commands[cid]
return 404, {"error": "not_found", "detail": path}
- Step 2: Тесты refresh
tests/test_refresh.py:
import unittest
from _common import SETTINGS, agent, jira_cfg, no_sleep
from fake_hub import FakeHub
from fake_jira import FakeJira
class RefreshTest(unittest.TestCase):
def setUp(self):
self.fake = FakeJira()
self.hub = FakeHub(SETTINGS)
cfg = jira_cfg(self.fake, hub_url=self.hub.base_url, hub_token=self.hub.token)
self.jira = agent.JiraClient(cfg)
self.jira.sleep = no_sleep
self.ex = agent.Executor(self.jira, dict(SETTINGS), agent.HubClient(cfg))
self.ex.sleep = no_sleep
def tearDown(self):
self.fake.stop()
self.hub.stop()
def test_to_state_and_links(self):
self.fake.add("AURORA-1", assignee="ivanov", customfield_10801="2026-10-20T00:00:00.000+0300",
customfield_15302="AURORA-1350", labels=["AWPCA", "kind-dev"], sprint="Спринт 2")
self.fake.add("AURORA-2")
self.fake.links.append({"id": 1, "type": "ASUPR_Блокер_Начала", "source": "AURORA-1", "target": "AURORA-2"})
raw = self.fake.issue_json("AURORA-1")
st = self.ex.to_state(raw)
self.assertEqual(st["key"], "AURORA-1")
self.assertEqual(st["assignee"], "ivanov")
self.assertEqual(st["planned_start"], "2026-10-20")
self.assertEqual(st["epic_key"], "AURORA-1350")
self.assertEqual(st["labels"], ["AWPCA", "kind-dev"])
self.assertEqual(st["issue_type"], "Задача")
self.assertEqual(st["jira_updated"], "2026-10-01T10:00:00.000+0300")
self.assertIsNone(st["sprint"]) # спринт заполняет do_refresh
self.assertEqual(self.ex.links_from(raw), [{"source_key": "AURORA-1", "target_key": "AURORA-2", "link_type": "ASUPR_Блокер_Начала"}])
self.assertEqual(self.ex.links_from(self.fake.issue_json("AURORA-2")),
[{"source_key": "AURORA-1", "target_key": "AURORA-2", "link_type": "ASUPR_Блокер_Начала"}])
self.assertIn("customfield_15302", self.ex.state_fields())
def test_full_refresh_with_neighbors_and_sprints(self):
for i in range(1, 4):
self.fake.add("AURORA-%d" % i, sprint="Спринт 2" if i < 3 else None)
self.fake.add("OTHER-9", in_base=False)
self.fake.add("AURORA-7", in_base=False)
self.fake.links.append({"id": 1, "type": "ASUPR_Связанные", "source": "AURORA-1", "target": "OTHER-9"})
self.fake.links.append({"id": 2, "type": "ASUPR_Связанные", "source": "AURORA-7", "target": "AURORA-2"})
self.hub.known = ["AURORA-7"]
out = self.ex.run({"id": 5, "type": "refresh", "params": {"jql": "project = AURORA AND labels = AWPCA"}})
self.assertEqual(out["status"], "done")
self.assertEqual(out["result"], {"issues": 3, "neighbors": 1, "pages": 1})
state = self.hub.states[0]
self.assertEqual(state["command_id"], 5)
self.assertEqual(sorted(i["key"] for i in state["issues"]), ["AURORA-1", "AURORA-2", "AURORA-3", "OTHER-9"])
self.assertEqual(state["neighbor_keys"], ["OTHER-9"])
self.assertTrue(state["last_page"])
sprints = {i["key"]: i["sprint"] for i in state["issues"]}
self.assertEqual(sprints["AURORA-1"], "Спринт 2")
self.assertIsNone(sprints["AURORA-3"])
links = {(l["source_key"], l["target_key"]) for l in state["links"]}
self.assertEqual(links, {("AURORA-1", "OTHER-9"), ("AURORA-7", "AURORA-2")})
self.assertEqual([r[1] for r in self.fake.requests if r[1] == "/rest/api/2/search"].count("/rest/api/2/search"), 2)
def test_refresh_by_keys_pages(self):
for i in range(1, 121):
self.fake.add("AURORA-%d" % i)
keys = ["AURORA-%d" % i for i in range(1, 121)]
out = self.ex.run({"id": 6, "type": "refresh", "params": {"keys": keys}})
self.assertEqual(out["result"], {"issues": 120, "neighbors": 0, "pages": 2})
self.assertEqual(len(self.hub.states), 2)
self.assertEqual(self.hub.states[0]["page"], 1)
self.assertFalse(self.hub.states[0]["last_page"])
self.assertTrue(self.hub.states[1]["last_page"])
self.assertEqual(len(self.hub.states[0]["issues"]) + len(self.hub.states[1]["issues"]), 120)
def test_refresh_empty_keys(self):
out = self.ex.run({"id": 7, "type": "refresh", "params": {"keys": []}})
self.assertEqual(out["result"], {"issues": 0, "neighbors": 0, "pages": 0})
self.assertEqual(self.fake.requests, [])
self.assertEqual(self.hub.states, [])
if __name__ == "__main__":
unittest.main()
- Step 3: Запустить — падает
Run: /usr/bin/python3 -m unittest tests.test_refresh -v
Expected: FAIL — AttributeError: 'Executor' object has no attribute 'to_state'
- Step 4: Реализация
В класс Executor после do_ensure_sprint:
# --- refresh ---------------------------------------------------------------
STATE_FIELDS = ["summary", "description", "status", "priority", "assignee", "reporter", "duedate", "labels",
"issuetype", "created", "updated", "issuelinks"]
def state_fields(self) -> str:
extra = [self.fid(x) for x in ("planned_start", "planned_end") if self.has_field(x)]
ep = (self.s.get("epic_link") or {})
if ep.get("mode", "field") == "field" and ep.get("field_id"):
extra.append(ep["field_id"])
return ",".join(self.STATE_FIELDS + extra)
def search_all(self, jql: str) -> List[dict]:
issues, start = [], 0
while True:
page = self.jira.get(API + "/search", jql=jql, fields=self.state_fields(), startAt=start, maxResults=100)
got = page.get("issues") or []
issues += got
start += len(got)
if not got or start >= int(page.get("total") or 0):
return issues
def to_state(self, raw: dict) -> dict:
f = raw.get("fields") or {}
def name(obj):
return (obj or {}).get("name")
ep = self.s.get("epic_link") or {}
epic = f.get(ep["field_id"]) if ep.get("mode", "field") == "field" and ep.get("field_id") else None
return {"key": raw["key"], "id": raw.get("id"), "issue_type": name(f.get("issuetype")), "summary": f.get("summary"),
"description": f.get("description"), "status": name(f.get("status")), "priority": name(f.get("priority")),
"assignee": name(f.get("assignee")), "reporter": name(f.get("reporter")), "duedate": _date(f.get("duedate")),
"planned_start": _date(f.get(self.fid("planned_start"))) if self.has_field("planned_start") else None,
"planned_end": _date(f.get(self.fid("planned_end"))) if self.has_field("planned_end") else None,
"epic_key": epic, "labels": list(f.get("labels") or []), "sprint": None,
"created": f.get("created"), "jira_updated": f.get("updated")}
def links_from(self, raw: dict) -> List[dict]:
key, out = raw["key"], []
for l in (raw.get("fields") or {}).get("issuelinks") or []:
t = (l.get("type") or {}).get("name")
if l.get("outwardIssue"):
out.append({"source_key": key, "target_key": l["outwardIssue"]["key"], "link_type": t})
if l.get("inwardIssue"):
out.append({"source_key": l["inwardIssue"]["key"], "target_key": key, "link_type": t})
return out
def sprint_names(self, keys: List[str]) -> Dict[str, str]:
names = {}
if not keys or not self.s.get("board_id"):
return names
for name, sp in self.sprints().items():
for i in range(0, len(keys), 100):
chunk = keys[i:i + 100]
page = self.jira.get(AGILE + "/sprint/%s/issue" % sp["id"], jql="key in (%s)" % ",".join(chunk),
fields="key", maxResults=200)
for it in page.get("issues") or []:
names[it["key"]] = name
return names
def do_refresh(self, p: dict, cmd: dict) -> dict:
if p.get("jql"):
raws = self.search_all(p["jql"])
else:
keys = [k for k in (p.get("keys") or []) if k]
if not keys:
return self._out("done", {"issues": 0, "neighbors": 0, "pages": 0})
raws = []
for i in range(0, len(keys), 100):
raws += self.search_all("key in (%s)" % ",".join(keys[i:i + 100]))
fetched = {r["key"] for r in raws}
known = set(self.hub.known_keys()) if self.hub else set()
linked = set()
for r in raws:
for l in self.links_from(r):
linked.add(l["source_key"])
linked.add(l["target_key"])
missing = sorted(k for k in linked if k not in fetched and k not in known)
neighbors = []
for i in range(0, len(missing), 100):
neighbors += self.search_all("key in (%s)" % ",".join(missing[i:i + 100]))
everything = raws + neighbors
names = self.sprint_names([r["key"] for r in everything])
issues, links, seen = [], [], set()
for r in everything:
st = self.to_state(r)
st["sprint"] = names.get(r["key"])
issues.append(st)
for l in self.links_from(r):
sig = (l["source_key"], l["target_key"], l["link_type"])
if sig not in seen:
seen.add(sig)
links.append(l)
neighbor_keys = [r["key"] for r in neighbors]
pages = 0
for i in range(0, len(issues), 100):
chunk = issues[i:i + 100]
chunk_keys = {x["key"] for x in chunk}
pages += 1
self.hub.state({"command_id": cmd.get("id"), "issues": chunk,
"links": [l for l in links if l["source_key"] in chunk_keys or l["target_key"] in chunk_keys],
"neighbor_keys": [k for k in neighbor_keys if k in chunk_keys],
"page": pages, "last_page": i + 100 >= len(issues)})
return self._out("done", {"issues": len(raws), "neighbors": len(neighbors), "pages": pages})
- Step 5: Запустить — зелёный
Run: /usr/bin/python3 -m unittest tests.test_refresh -v
Expected: 4 passed
- Step 6: Коммит
git add src/agent.py tests/fake_hub.py tests/test_refresh.py
git commit -m "feat(agent): чтение состояния — поиск, соседи по связям, спринты, страницы в хаб
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>"
Task 6: Цикл агента, восстановление, команда run
Files:
- Modify:
src/agent.py(классRunnerпослеExecutor,cmd_run,main) - Test:
tests/test_runner.py
Interfaces:
-
Produces:
Runner(cfg, hub, jira, journal):.settings,.executor,.stop(флаг),.sleep(подменяемый),.idle_limit(None | int — выйти после N пустых опросов, для тестов),.load_settings(),.send_result(rec) -> bool,.wait_pause(pause_seconds) -> bool,.execute(cmd),.recover(),.loop();cmd_run(cfg, journal_path) -> int. -
Step 1: Тесты
tests/test_runner.py:
import os
import tempfile
import time
import unittest
from _common import SETTINGS, agent, jira_cfg, no_sleep
from fake_hub import FakeHub
from fake_jira import FakeJira
class RunnerTest(unittest.TestCase):
def setUp(self):
self.fake = FakeJira()
self.hub = FakeHub(SETTINGS)
self.dir = tempfile.TemporaryDirectory()
self.journal = agent.Journal(os.path.join(self.dir.name, "journal.jsonl"))
cfg = jira_cfg(self.fake, hub_url=self.hub.base_url, hub_token=self.hub.token)
self.runner = agent.Runner(cfg, agent.HubClient(cfg), agent.JiraClient(cfg), self.journal)
self.sleeps = []
self.runner.sleep = self.sleeps.append
self.runner.jira.sleep = no_sleep
self.runner.idle_limit = 1
self.fake.add("AURORA-1", duedate="2026-10-20")
def tearDown(self):
self.fake.stop()
self.hub.stop()
self.dir.cleanup()
def kinds(self):
return [(r["kind"], r["command_id"]) for r in self.journal.records()]
def test_processes_queue(self):
self.hub.enqueue("ensure_fields", {"key": "AURORA-1", "duedate": "2026-10-23"})
self.hub.enqueue("refresh", {"keys": ["AURORA-1"]})
self.runner.loop()
self.assertEqual(self.hub.results[1]["status"], "done")
self.assertEqual(self.hub.results[1]["result"], {"changed": ["duedate"]})
self.assertEqual(self.hub.results[1]["jira_calls"], 2)
self.assertEqual(self.hub.results[2]["status"], "done")
self.assertEqual(len(self.hub.states), 1)
self.assertEqual(self.hub.states[0]["issues"][0]["duedate"], "2026-10-23")
self.assertEqual(self.kinds(), [("started", 1), ("finished", 1), ("acked", 1), ("started", 2), ("finished", 2), ("acked", 2)])
self.assertEqual(self.sleeps, []) # пауз не было: last_call_at пуст, потом pause 0
def test_pause_waits_from_last_jira_call(self):
self.runner.jira.last_call_at = time.monotonic()
self.hub.enqueue("ensure_fields", {"key": "AURORA-1", "duedate": "2026-10-23"}, pause=3)
self.runner.loop()
self.assertEqual(len(self.sleeps), 3)
self.assertAlmostEqual(sum(self.sleeps), 3, delta=0.1)
self.assertEqual(self.hub.results[1]["status"], "done")
def test_result_retried_when_hub_down(self):
self.hub.fail_results = 2
self.hub.enqueue("ensure_fields", {"key": "AURORA-1", "duedate": "2026-10-23"})
self.runner.loop()
self.assertEqual(self.hub.results[1]["status"], "done")
self.assertEqual(self.sleeps, [10, 10])
self.assertEqual(self.journal.unacked_finished(), [])
def test_result_rejected_is_acked(self):
self.hub.reject_results = {1: 404}
self.hub.enqueue("ensure_fields", {"key": "AURORA-1", "duedate": "2026-10-23"})
self.runner.loop()
self.assertNotIn(1, self.hub.results)
self.assertEqual(self.journal.unacked_finished(), [])
self.assertEqual(self.kinds()[-1], ("acked", 1))
def test_recover_unfinished_started(self):
cid = self.hub.enqueue("ensure_fields", {"key": "AURORA-1", "duedate": "2026-10-20"}, pause=99)
self.hub.queue.remove(cid)
self.hub.commands[cid]["status"] = "claimed"
self.hub.in_flight = cid
self.journal.started(cid)
self.runner.loop()
self.assertEqual(self.hub.results[cid]["result"], {"changed": []}) # уже так в Jira — записи нет
self.assertEqual(self.journal.unfinished_started(), [])
self.assertEqual(self.sleeps, []) # пауза при восстановлении не применяется
self.assertEqual([r[0] for r in self.fake.writes()], [])
def test_recover_skips_command_not_claimed_anymore(self):
cid = self.hub.enqueue("ensure_fields", {"key": "AURORA-1", "duedate": "2026-10-23"})
self.hub.queue.remove(cid)
self.hub.commands[cid]["status"] = "cancelled"
self.journal.started(cid)
self.runner.loop()
self.assertNotIn(cid, self.hub.results)
self.assertEqual(self.kinds(), [("started", cid), ("acked", cid)])
def test_recover_unacked_finished(self):
cid = self.hub.enqueue("ensure_fields", {"key": "AURORA-1", "duedate": "2026-10-23"})
self.hub.queue.remove(cid)
self.hub.commands[cid]["status"] = "claimed"
self.hub.in_flight = cid
self.journal.started(cid)
self.journal.finished(cid, "done", {"changed": ["duedate"]}, None, 2, 0)
self.hub.enqueue("refresh", {"keys": ["AURORA-1"]})
self.runner.loop()
self.assertEqual(self.hub.results[cid]["result"], {"changed": ["duedate"]})
self.assertEqual(self.hub.results[2]["status"], "done")
self.assertEqual(self.kinds()[:3], [("started", 1), ("finished", 1), ("acked", 1)])
def test_stop_during_pause(self):
self.runner.jira.last_call_at = time.monotonic()
self.hub.enqueue("ensure_fields", {"key": "AURORA-1", "duedate": "2026-10-23"}, pause=5)
self.runner.sleep = lambda s: setattr(self.runner, "stop", True)
self.runner.loop()
self.assertEqual(self.journal.records(), [])
self.assertEqual(self.hub.results, {})
self.assertEqual(self.hub.commands[1]["status"], "claimed")
def test_settings_reload_on_version_change(self):
self.runner.load_settings()
self.runner.settings["settings_version"] = 0
self.runner.loop()
self.assertEqual(self.runner.settings["settings_version"], 1)
def test_hub_down_at_start_keeps_polling(self):
calls = []
def hub_settings():
calls.append(1)
if len(calls) < 3:
raise ConnectionError("нет связи")
return dict(SETTINGS)
self.runner.hub.settings = hub_settings
self.runner.loop()
self.assertEqual(len(calls), 3)
self.assertEqual(self.sleeps, [0, 0])
if __name__ == "__main__":
unittest.main()
- Step 2: Запустить — падает
Run: /usr/bin/python3 -m unittest tests.test_runner -v
Expected: FAIL — AttributeError: module 'agent' has no attribute 'Runner'
- Step 3: Реализация
В src/agent.py после класса Executor:
# ---------------------------------------------------------------------------
# Цикл: claim → пауза → журнал → выполнение → отчёт до подтверждения; восстановление при старте
# ---------------------------------------------------------------------------
class Runner:
def __init__(self, cfg: dict, hub: HubClient, jira: JiraClient, journal: Journal):
self.cfg, self.hub, self.jira, self.journal = cfg, hub, jira, journal
self.settings = None # type: Optional[dict]
self.executor = None # type: Optional[Executor]
self.stop = False
self.sleep = time.sleep
self.idle_limit = None # type: Optional[int]
@property
def poll(self) -> float:
return max(float(self.cfg.get("poll_seconds", 5) or 0), 0)
def load_settings(self) -> None:
self.settings = self.hub.settings()
self.executor = Executor(self.jira, self.settings, self.hub)
self.executor.sleep = self.sleep
log("настройки хаба v%s (проект %s)" % (self.settings.get("settings_version"), self.settings.get("project_key")))
def send_result(self, rec: dict) -> bool:
"""Отчёт хабу; повторяется, пока хаб не подтвердит. 404/409/422 — хаб отверг (команда отменена или удалена):
считаем закрытой, чтобы не зациклиться."""
cid = rec["command_id"]
payload = {"status": rec["status"], "result": rec.get("result"), "error": rec.get("error"),
"jira_calls": rec.get("jira_calls", 0), "rate_limited": rec.get("rate_limited", 0)}
while not self.stop:
try:
self.hub.result(cid, payload)
self.journal.acked(cid)
return True
except HttpError as e:
if e.code in (404, 409, 422):
log("хаб отверг отчёт по #%s: %s — считаю закрытым" % (cid, e))
self.journal.acked(cid)
return True
log("хаб: %s, повтор через 10 с" % e)
except ConnectionError as e:
log("хаб недоступен: %s, повтор через 10 с" % e)
self.sleep(10)
return False
def wait_pause(self, pause_seconds) -> bool:
"""Досыпать до зазора от последнего вызова Jira; False — остановлены во время паузы."""
pause = float(pause_seconds or 0)
if pause > 0 and self.jira.last_call_at is not None:
remaining = pause - (time.monotonic() - self.jira.last_call_at)
if remaining > 0:
log(" пауза %.0f с до команды" % remaining)
while remaining > 0 and not self.stop:
step = min(remaining, 1.0)
self.sleep(step)
remaining -= step
return not self.stop
def execute(self, cmd: dict) -> None:
log("#%s %s %s" % (cmd["id"], cmd["type"], json.dumps(cmd.get("params"), ensure_ascii=False)[:200]))
if not self.wait_pause(cmd.get("pause_seconds")):
return
self.journal.started(cmd["id"])
out = self.executor.run(cmd)
self.journal.finished(cmd["id"], out["status"], out.get("result"), out.get("error"), out["jira_calls"], out["rate_limited"])
log(" → %s %s" % (out["status"], out.get("error") or json.dumps(out.get("result"), ensure_ascii=False)))
rec = dict(out)
rec["command_id"] = cmd["id"]
self.send_result(rec)
def recover(self) -> None:
for rec in self.journal.unacked_finished():
log("досылаю отчёт по #%s" % rec["command_id"])
self.send_result(rec)
for cid in self.journal.unfinished_started():
try:
cmd = self.hub.command(cid)
except HttpError as e:
log("#%s: хаб ответил %s — закрываю запись журнала" % (cid, e))
self.journal.acked(cid)
continue
if cmd.get("status") != "claimed":
log("#%s уже %s в хабе — пропускаю" % (cid, cmd.get("status")))
self.journal.acked(cid)
continue
log("#%s начата до перезапуска — выполняю заново через проверку" % cid)
cmd["pause_seconds"] = 0
self.execute(cmd)
def loop(self) -> None:
idle, recovered = 0, False
while not self.stop:
if self.settings is None:
try:
self.load_settings()
except (ConnectionError, HttpError) as e:
log("хаб: %s" % e)
self.sleep(self.poll)
continue
if not recovered:
self.recover()
recovered = True
try:
resp = self.hub.claim(AGENT_VERSION, self.settings.get("settings_version"), self.jira.last_call_iso)
except (ConnectionError, HttpError) as e:
log("хаб: %s" % e)
self.sleep(self.poll)
continue
if resp.get("settings_version") != self.settings.get("settings_version"):
try:
self.load_settings()
except (ConnectionError, HttpError) as e:
log("хаб: не перечитать настройки: %s" % e)
cmd = resp.get("command")
if cmd:
idle = 0
self.execute(cmd)
continue
idle += 1
if self.idle_limit is not None and idle >= self.idle_limit:
return
self.sleep(self.poll)
def cmd_run(cfg: dict, journal_path: str) -> int:
runner = Runner(cfg, HubClient(cfg), JiraClient(cfg), Journal(journal_path))
def on_signal(signum, frame):
if runner.stop:
log("повторный сигнал — выхожу немедленно")
sys.exit(130)
runner.stop = True
log("сигнал %s: доделываю текущую команду и выхожу" % signum)
signal.signal(signal.SIGINT, on_signal)
signal.signal(signal.SIGTERM, on_signal)
log("jira-agent v%d: хаб %s, Jira %s, журнал %s" % (AGENT_VERSION, cfg["hub_url"], cfg["jira"]["base_url"], journal_path))
runner.loop()
log("остановлен")
return 0
В main заменить хвост:
if args.cmd == "check":
return cmd_check(cfg)
if args.cmd == "prune":
removed = Journal(args.journal).prune(30)
log("журнал: удалено записей %d" % removed)
return 0
return cmd_run(cfg, args.journal)
- Step 4: Запустить — зелёный, затем весь набор
Run: /usr/bin/python3 -m unittest tests.test_runner -v && /usr/bin/python3 -m unittest discover -s tests -t . -v
Expected: 10 passed; весь набор зелёный
- Step 5: Коммит
git add src/agent.py tests/test_runner.py
git commit -m "feat(agent): цикл claim → пауза → журнал → отчёт, восстановление после перезапуска, run
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>"
Task 7: README, CLAUDE.md, ARCHITECTURE.md, проверка синтаксиса 3.9
Files:
-
Create:
README.md,tests/test_py39.py,.claude/code-paths(если хук require-plan не видитsrc/по умолчанию — проверитьcat .claude/hooks/*в репозитории; при стандартной обвязке файл не нужен) -
Modify:
CLAUDE.md,ARCHITECTURE.md -
Step 1: Тест синтаксиса
tests/test_py39.py:
import ast
import os
import unittest
from _common import SRC
class Py39Test(unittest.TestCase):
def test_agent_parses_as_python_39(self):
with open(os.path.join(SRC, "agent.py"), encoding="utf-8") as f:
ast.parse(f.read(), feature_version=(3, 9))
def test_no_third_party_imports(self):
with open(os.path.join(SRC, "agent.py"), encoding="utf-8") as f:
tree = ast.parse(f.read())
allowed = {"argparse", "base64", "json", "os", "signal", "ssl", "sys", "time", "urllib", "datetime", "typing"}
for node in ast.walk(tree):
if isinstance(node, ast.Import):
for a in node.names:
self.assertIn(a.name.split(".")[0], allowed, a.name)
elif isinstance(node, ast.ImportFrom):
self.assertIn((node.module or "").split(".")[0], allowed, node.module)
if __name__ == "__main__":
unittest.main()
Run: /usr/bin/python3 -m unittest tests.test_py39 -v
Expected: 2 passed
- Step 2: README.md
Написать полностью на русском по пунктам:
- Что это и откуда спека (
../jira-hub/openspec/changes/bridge-core/specs/agent/spec.md); связь сjira-hub. - Доставка на виртуалку: скопировать
src/agent.pyиsrc/agent.example.jsonчерез общую папку в каталог на виртуалке (например~/jira-agent/), переименовать вagent.json, заполнитьhub_url(http://192.168.0.106:8790),hub_token(изjira-hub tokensна Mac mini),jira.base_url,jira.auth(логин/пароль илиtoken),verify_ssl/ca_bundleпри корпоративном сертификате. Напоминание:agent.jsonникуда не копировать, кроме виртуалки. - Проверка:
python3 agent.py check— что печатает в норме и при ошибках. - Запуск:
python3 agent.pyв терминале; в фоне —nohup python3 -u agent.py > agent.log 2>&1 &; остановка — Ctrl-C илиkill <pid>(текущая команда доделывается, повторный сигнал — выход немедленно). - Автозапуск через launchd: пример
~/Library/LaunchAgents/ru.aepif.jira-agent.plist(ProgramArguments—/usr/bin/python3,-u, путь кagent.py;WorkingDirectory;RunAtLoad;KeepAlive;StandardOutPath/StandardErrorPath), командыlaunchctl load/unload. - Журнал:
journal.jsonlрядом с агентом — что в нём, зачем,python3 agent.py prune(подтверждённые записи старше 30 дней); журнал содержит параметры команд (тексты задач) — не выносить с виртуалки. - Что делает агент с командами: таблица типов со ссылкой на спеку; правило «проверь, потом сделай»; паузы приходят от хаба; 429/503 — Retry-After; версия агента
AGENT_VERSIONи предупреждение хаба об устаревшем агенте. - Тесты:
/usr/bin/python3 -m unittest discover -s tests -t .(тот же Python 3.9.6, что на виртуалке). - Обновление агента: скопировать новый
agent.py, перезапустить; журнал иagent.jsonне трогать.
- Step 3: CLAUDE.md и ARCHITECTURE.md
Заменить {{...}} шаблона: что это (абзац из proposal хаба про агента), стек (Python 3.9.6, только stdlib, unittest), архитектура —
один файл src/agent.py с разделами (HTTP-клиенты, Journal, Executor, Runner, CLI) и тесты с фейковыми Jira/хабом,
команды (/usr/bin/python3 -m unittest discover -s tests -t ., python3 src/agent.py check), тесты (фейки в tests/fake_*.py,
внешних сервисов нет), первая подсистема — исполнитель команд. В ARCHITECTURE.md — таблица разделов файла и границы:
«пароль Jira только в agent.json», «политика темпа не хранится», «одна команда в полёте», «отчёт до подтверждения».
Указать, что спека и план живут в jira-hub (openspec/changes/bridge-core, docs/superpowers/plans/2026-10-11-bridge-core-hub.md).
- Step 4: Весь набор и коммит
Run: /usr/bin/python3 -m unittest discover -s tests -t . -v
Expected: все зелёные
git add README.md CLAUDE.md ARCHITECTURE.md tests/test_py39.py
git commit -m "docs(agent): README (доставка, запуск, launchd, prune), архитектура, проверка синтаксиса 3.9
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>"