240 lines
11 KiB
Python
Executable file
240 lines
11 KiB
Python
Executable file
#!/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())
|