#!/usr/bin/env python3 """Шина сессий: захват зон и адресные сообщения между параллельными сессиями. Объект: дом агента (agent-home-kit). Назначение — развести одновременные работы по зонам, не сериализуя их. Захват объявляет зону и сразу отдаёт соседей-держателей и почту по этой зоне: договариваются сессии между собой, без прохода через оператора. Хранение append-only (state/sessions/bus.jsonl), свёртка при чтении. Живость держателя берётся из реестра сессий: молчит дольше часа — захват остыл (порог сироты session-reg.py), мёртвая сессия зону не держит. Режим по умолчанию — наблюдение: захват занятой зоны не запрещается, а показывается. """ from __future__ import annotations import argparse, datetime as dt, fcntl, json, os, tempfile from pathlib import Path import sys as _sys; _sys.path.insert(0, str(Path(__file__).resolve().parent)) from kit import HOME # корень дома из kit.json (HOME_KIT_ROOT) BUS = Path(os.environ.get("KIT_BUS", HOME / "state" / "sessions" / "bus.jsonl")) EVENTS = Path(os.environ.get("KIT_SESSION_EVENTS", HOME / "state" / "sessions" / "events.jsonl")) COLD_MIN = 60 MSK = dt.timezone(dt.timedelta(hours=3)) def now() -> dt.datetime: return dt.datetime.now(dt.timezone.utc) def parse_ts(v: str) -> dt.datetime: p = dt.datetime.fromisoformat(str(v).replace("Z", "+00:00")) return p.replace(tzinfo=dt.timezone.utc) if p.tzinfo is None else p.astimezone(dt.timezone.utc) def mins_since(ts: dt.datetime) -> float: return (now() - ts).total_seconds() / 60 def hhmm(ts: dt.datetime) -> str: return ts.astimezone(MSK).strftime("%d.%m %H:%M мск") def append(rec: dict) -> dict: BUS.parent.mkdir(parents=True, exist_ok=True) rec = {"ts": now().isoformat(timespec="seconds"), **rec} with BUS.open("a", encoding="utf-8") as f: fcntl.flock(f, fcntl.LOCK_EX) f.write(json.dumps(rec, ensure_ascii=False) + "\n") return rec def rows(path: Path) -> list[dict]: if not path.exists(): return [] out = [] for line in path.read_text(encoding="utf-8", errors="replace").splitlines(): line = line.strip() if not line: continue try: out.append(json.loads(line)) except ValueError: continue return out def last_seen_min(label: str): newest = None for path, key in ((EVENTS, "label"), (BUS, "session")): for r in rows(path): if r.get(key) != label: continue try: ts = parse_ts(r["ts"]) except Exception: continue if newest is None or ts > newest: newest = ts return None if newest is None else mins_since(newest) def held() -> dict: cur: dict[str, dict] = {} for r in rows(BUS): if r.get("kind") == "claim": cur[r.get("zone", "")] = r elif r.get("kind") == "release": cur.pop(r.get("zone", ""), None) cur.pop("", None) return cur def holder_state(rec: dict) -> str: m = last_seen_min(rec.get("session", "")) if m is None: return "молчит" return "остыл" if m > COLD_MIN else "жив %d мин назад" % int(m) def zone_match(zone: str, other: str) -> bool: """пересечение зон: равенство или вложенность по двоеточию""" return zone == other or zone.startswith(other + ":") or other.startswith(zone + ":") def inbox_rows(session: str, zones: list[str]) -> list[dict]: seen = set() for r in rows(BUS): if r.get("session") != session: continue if r.get("kind") == "ack": seen.add(r.get("msg")) elif r.get("kind") == "ack_batch": seen.update(r.get("msgs") or []) out = [] for r in rows(BUS): if r.get("kind") != "msg" or r.get("id") in seen or r.get("session") == session: continue to = r.get("to", "") if to == session or to == "all" or (to.startswith("zone:") and any(zone_match(to[5:], z) for z in zones)): out.append(r) return out def my_zones(session: str) -> list[str]: return [z for z, r in held().items() if r.get("session") == session] def print_neighbours(zone: str, session: str) -> int: n = 0 for z, r in sorted(held().items()): if r.get("session") == session or not zone_match(zone, z): continue n += 1 print(" сосед: %s держит %s с %s (%s) — %s" % (r.get("session"), z, hhmm(parse_ts(r["ts"])), holder_state(r), r.get("why", ""))) return n def print_mail(session: str, zones: list[str]) -> int: msgs = inbox_rows(session, zones) for m in msgs: print(" письмо от %s (%s, %s): %s" % (m.get("session"), m.get("to"), hhmm(parse_ts(m["ts"])), m.get("text", ""))) append({"kind": "ack", "session": session, "msg": m.get("id")}) return len(msgs) def cmd_claim(a) -> int: others = print_neighbours(a.zone, a.session) rec = append({"kind": "claim", "zone": a.zone, "session": a.session, "why": a.why or ""}) print("захват: %s ← %s" % (a.zone, a.session)) mail = print_mail(a.session, my_zones(a.session) + [a.zone]) if not others and not mail: print(" зона свободна, писем нет") if others: print(" ДЕЙСТВИЕ: зона занята — напиши держателю (session-bus.py send --to <метка>) или возьми соседнюю; отказа нет, наслоение на тебе") return 0 def cmd_release(a) -> int: if a.zone not in held(): print("нечего отпускать: %s не занята" % a.zone) return 0 append({"kind": "release", "zone": a.zone, "session": a.session}) print("отпущено: %s" % a.zone) return 0 def cmd_zones(a) -> int: cur = held() if a.zone: cur = {z: r for z, r in cur.items() if zone_match(a.zone, z)} if not cur: print("захватов нет") return 0 for z, r in sorted(cur.items()): print("%-34s %-28s %s %s" % (z, r.get("session"), holder_state(r), r.get("why", ""))) return 0 def cmd_send(a) -> int: rec = append({"kind": "msg", "id": os.urandom(6).hex(), "to": a.to, "session": a.session, "text": a.text}) print("отправлено %s → %s" % (rec["id"], a.to)) return 0 def cmd_inbox(a) -> int: zones = my_zones(a.session) + (a.zone or []) n = print_mail(a.session, zones) if not n: print("писем нет") return 0 def cmd_selftest(a) -> int: global BUS, EVENTS with tempfile.TemporaryDirectory() as d: BUS = Path(d) / "bus.jsonl" EVENTS = Path(d) / "events.jsonl" EVENTS.write_text(json.dumps({"ts": now().isoformat(timespec="seconds"), "kind": "open", "label": "s-a"}) + "\n", encoding="utf-8") append({"kind": "claim", "zone": "mesto:scene", "session": "s-a", "why": "музей"}) assert "mesto:scene" in held(), "захват не виден" assert holder_state(held()["mesto:scene"]).startswith("жив"), "живость держателя не читается" append({"kind": "msg", "id": "m1", "to": "zone:mesto:scene", "session": "s-a", "text": "беру этаж 2"}) assert [m["id"] for m in inbox_rows("s-b", ["mesto:scene:hall2"])] == ["m1"], "письмо зоне не дошло по вложенности" assert inbox_rows("s-a", ["mesto:scene"]) == [], "автор получил своё письмо" append({"kind": "ack", "session": "s-b", "msg": "m1"}) assert inbox_rows("s-b", ["mesto:scene"]) == [], "письмо пришло повторно после ack" append({"kind": "msg", "id": "m2", "to": "zone:mesto:scene", "session": "s-a", "text": "пачка"}) append({"kind": "ack_batch", "session": "s-b", "msgs": ["m2"], "summary": "свёртка"}) assert inbox_rows("s-b", ["mesto:scene"]) == [], "письмо пришло повторно после ack_batch" append({"kind": "release", "zone": "mesto:scene", "session": "s-a"}) assert held() == {}, "отпускание не сняло захват" old = now() - dt.timedelta(minutes=COLD_MIN + 30) EVENTS.write_text(json.dumps({"ts": old.isoformat(timespec="seconds"), "kind": "open", "label": "s-c"}) + "\n", encoding="utf-8") with BUS.open("a", encoding="utf-8") as f: f.write(json.dumps({"ts": old.isoformat(timespec="seconds"), "kind": "claim", "zone": "mesto:board", "session": "s-c"}) + "\n") assert holder_state(held()["mesto:board"]) == "остыл", "мёртвая сессия держит зону" print("PASS захват, вложенность зон, письма, ack, остывание") return 0 def main() -> int: p = argparse.ArgumentParser(description="шина сессий: зоны и письма между параллельными сессиями") s = p.add_subparsers(dest="cmd", required=True) c = s.add_parser("claim", help="объявить зону; печатает соседей и почту по зоне") c.add_argument("zone"); c.add_argument("--session", required=True); c.add_argument("--why", default="") c.set_defaults(fn=cmd_claim) r = s.add_parser("release", help="отпустить зону") r.add_argument("zone"); r.add_argument("--session", required=True); r.set_defaults(fn=cmd_release) z = s.add_parser("zones", help="кто что держит") z.add_argument("--zone"); z.set_defaults(fn=cmd_zones) m = s.add_parser("send", help="письмо сессии (--to метка) или зоне (--to zone:<зона>) или всем (--to all)") m.add_argument("text"); m.add_argument("--to", required=True); m.add_argument("--session", required=True) m.set_defaults(fn=cmd_send) i = s.add_parser("inbox", help="забрать письма (помечает доставленными)") i.add_argument("--session", required=True); i.add_argument("--zone", action="append"); i.set_defaults(fn=cmd_inbox) t = s.add_parser("selftest", help="проба прибора на временном хранилище"); t.set_defaults(fn=cmd_selftest) a = p.parse_args() return a.fn(a) if __name__ == "__main__": raise SystemExit(main())