Coverage for astrocyte/harness/team.py: 88%

1"""``astrocyte team`` — share a project's memory through the team's gateway. 

2 

3The client side of team memory (``docs/_design/team-memory.md``, C1): manual 

4push and pull, with a preview before anything leaves the machine. Hooks and 

5the MCP server keep reading and writing only the local store; this module 

6copies between it and the gateway's bank of the same id. 

7 

8- **Push** sends the project's shareable memories (``/sync/push``), each as 

9 one row under its own id: what an agent saved and what was imported, never 

10 captured conversation turns (unless the project opts in) or anything tagged 

11 ``private``. Teammates' memories pulled here are never pushed back. 

12- **Pull** pages the bank's changes feed (``/changes``) from a saved cursor: 

13 a teammate's memory is stored locally under the same id, through the same 

14 policy layer as a push, keeping who saved it (``_actor``); a tombstone 

15 erases that id here, from disk, including a memory of yours that a 

16 teammate forgot for the team. 

17 

18Membership (gateway URL per project bank) lives in ``team.json`` beside the 

19config; the token goes to the OS keychain when ``keyring`` is installed, else 

20into that file (0600). Sync state (the push ledger, the ids pulled, the feed 

21cursor) lives in the state directory, never in the memory store, so 

22``astrocyte memory export`` stays clean. 

23""" 

24 

25from __future__ import annotations 

26 

27import json 

28import os 

29import sys 

30from argparse import Namespace 

31from dataclasses import dataclass, field 

32from datetime import datetime, timezone 

33from pathlib import Path 

34from typing import Any 

35from urllib.parse import quote 

36 

37from .paths import config_path, state_dir 

38 

39KEYRING_SERVICE = "astrocyte-team" 

40PUSH_BATCH = 100 # the gateway's limit per request 

41PULL_PAGE = 500 

42PREVIEW_SAMPLE = 3 

43#: Metadata a push carries. Other underscore keys are the store's own, and the 

44#: gateway would drop them anyway. 

45_PUSHED_SYSTEM_KEYS = ("_created_at", "_retain_id", "_chunk_index") 

46 

47 

48class TeamError(Exception): 

49 """A problem the user can act on; printed without a traceback.""" 

50 

51 

52# ── membership and token ───────────────────────────────────────────────── 

53 

54 

55def team_file(cfg_path: Path) -> Path: 

56 return cfg_path.parent / "team.json" 

57 

58 

59def load_memberships(cfg_path: Path) -> dict[str, dict[str, Any]]: 

60 """Bank id → ``{"url", "share_captured", "joined_at", "token"?}``.""" 

61 try: 

62 data = json.loads(team_file(cfg_path).read_text(encoding="utf-8")) 

63 except (OSError, ValueError): 

64 return {} 

65 projects = data.get("projects") if isinstance(data, dict) else None 

66 return {k: v for k, v in projects.items() if isinstance(v, dict)} if isinstance(projects, dict) else {} 

67 

68 

69def save_memberships(cfg_path: Path, projects: dict[str, dict[str, Any]]) -> None: 

70 path = team_file(cfg_path) 

71 if not projects: 

72 path.unlink(missing_ok=True) 

73 return 

74 path.parent.mkdir(parents=True, exist_ok=True) 

75 tmp = path.with_name(path.name + ".tmp") 

76 # Created private: it may hold a token. 

77 fd = os.open(tmp, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600) 

78 with os.fdopen(fd, "w", encoding="utf-8") as f: 

79 json.dump({"projects": projects}, f, indent=2) 

80 os.replace(tmp, path) 

81 

82 

83def _keyring() -> Any | None: 

84 try: 

85 import keyring 

86 except ImportError: 

87 return None 

88 return keyring 

89 

90 

91def _keychain_account(bank: str, url: str) -> str: 

92 # Per project and gateway: one person may hold differently scoped tokens 

93 # for two projects on one gateway, and leaving one must not drop the other. 

94 return f"{bank}@{url}" 

95 

96 

97def store_token(bank: str, url: str, token: str, entry: dict[str, Any]) -> str: 

98 """Keep ``token`` for this project's gateway: the OS keychain if 

99 available, else the membership entry (written 0600). Returns where.""" 

100 kr = _keyring() 

101 if kr is not None: 

102 account = _keychain_account(bank, url) 

103 try: 

104 kr.set_password(KEYRING_SERVICE, account, token) 

105 # Read it back: some backends accept a password and keep nothing 

106 # (the null backend, a locked or broken store). 

107 if kr.get_password(KEYRING_SERVICE, account) == token: 

108 entry.pop("token", None) 

109 return "the OS keychain" 

110 except Exception: # noqa: BLE001 — no usable backend (headless Linux, CI) 

111 pass 

112 entry["token"] = token 

113 return "team.json (readable only by you)" 

114 

115 

116def token_for(bank: str, entry: dict[str, Any]) -> str | None: 

117 if isinstance(entry.get("token"), str): 

118 return entry["token"] 

119 kr = _keyring() 

120 if kr is None: 

121 return None 

122 try: 

123 return kr.get_password(KEYRING_SERVICE, _keychain_account(bank, entry["url"])) 

124 except Exception: # noqa: BLE001 

125 return None 

126 

127 

128def forget_token(bank: str, url: str) -> None: 

129 kr = _keyring() 

130 if kr is None: 

131 return 

132 try: 

133 kr.delete_password(KEYRING_SERVICE, _keychain_account(bank, url)) 

134 except Exception: # noqa: BLE001 — not there, or no backend 

135 pass 

136 

137 

138# ── sync state ─────────────────────────────────────────────────────────── 

139 

140 

141@dataclass 

142class SyncState: 

143 """Per bank: what each pushed id became, which ids came from teammates, 

144 and where the changes feed was left.""" 

145 

146 cursor: str | None = None 

147 #: id → "stored" | "unchanged" | "duplicate:<id>" | "rejected:<reason>" | "forgotten" 

148 pushed: dict[str, str] = field(default_factory=dict) 

149 pulled: list[str] = field(default_factory=list) 

150 #: ids erased here with ``forget --local``: a later pull must not bring them back 

151 suppressed: list[str] = field(default_factory=list) 

152 #: ids shared with ``memory share`` that the rules would keep here (a captured turn, ``private``) 

153 promoted: list[str] = field(default_factory=list) 

154 #: ids kept here with ``memory unshare``: never pushed, and their tombstone doesn't erase them here 

155 withheld: list[str] = field(default_factory=list) 

156 last_sync: str | None = None 

157 last_error: str | None = None 

158 

159 @staticmethod 

160 def path(bank: str) -> Path: 

161 safe = "".join(c if c.isalnum() or c in "._-" else "_" for c in bank) 

162 return state_dir() / "team" / f"{safe}.json" 

163 

164 @classmethod 

165 def load(cls, bank: str) -> SyncState: 

166 try: 

167 data = json.loads(cls.path(bank).read_text(encoding="utf-8")) 

168 return cls(cursor=data.get("cursor"), pushed=dict(data.get("pushed") or {}), 

169 pulled=list(data.get("pulled") or []), suppressed=list(data.get("suppressed") or []), 

170 promoted=list(data.get("promoted") or []), withheld=list(data.get("withheld") or []), 

171 last_sync=data.get("last_sync"), last_error=data.get("last_error")) 

172 except (OSError, ValueError, AttributeError, TypeError): 

173 return cls() 

174 

175 def save(self, bank: str) -> None: 

176 path = self.path(bank) 

177 path.parent.mkdir(parents=True, exist_ok=True, mode=0o700) 

178 tmp = path.with_name(path.name + ".tmp") 

179 tmp.write_text(json.dumps({"cursor": self.cursor, "pushed": self.pushed, "pulled": self.pulled, 

180 "suppressed": self.suppressed, "promoted": self.promoted, 

181 "withheld": self.withheld, "last_sync": self.last_sync, 

182 "last_error": self.last_error}), encoding="utf-8") 

183 os.replace(tmp, path) 

184 

185 def shared(self, memory_id: str) -> bool: 

186 """Is this id on the gateway: pulled from a teammate, or pushed and kept?""" 

187 return memory_id in self.pulled or self.pushed.get(memory_id) in ("stored", "unchanged") 

188 

189 

190def pulled_ids(bank: str) -> set[str]: 

191 """Ids of teammates' memories in ``bank`` on this machine (empty when the 

192 project isn't shared).""" 

193 return set(SyncState.load(bank).pulled) 

194 

195 

196class SyncBusy(TeamError): 

197 """Another process (the agent daemon, or a second `astrocyte team`) is syncing this bank.""" 

198 

199 

200class _BankLock: 

201 """One sync per bank at a time across processes: the daemon's background 

202 sync and a manual `astrocyte team sync` would otherwise interleave pages 

203 of the feed and both write the ledger.""" 

204 

205 def __init__(self, bank: str) -> None: 

206 self.path = SyncState.path(bank).with_suffix(".lock") 

207 self.fh: Any = None 

208 

209 def __enter__(self) -> _BankLock: 

210 from .agentd import _lock_exclusively 

211 

212 self.path.parent.mkdir(parents=True, exist_ok=True, mode=0o700) 

213 self.fh = open(self.path, "w") # noqa: SIM115 — held until __exit__ 

214 if not _lock_exclusively(self.fh): 

215 self.fh.close() 

216 raise SyncBusy("another sync of this project is running (the agent daemon syncs in the background)") 

217 return self 

218 

219 def __exit__(self, *exc: Any) -> None: 

220 self.fh.close() 

221 

222 

223# ── what is shared ─────────────────────────────────────────────────────── 

224 

225 

226def shareable(item: Any, state: SyncState, *, share_captured: bool) -> bool: 

227 """Would this local memory go to the team? Saved and imported memories 

228 do; captured turns only where the project opted in; ``private`` never; 

229 nor anything pulled from a teammate. ``memory share`` and ``unshare`` 

230 override the rules for one memory.""" 

231 from astrocyte._sync import PUSH_ID_RE 

232 

233 if item.id in state.withheld or item.id in state.pulled or (item.memory_layer or "fact") != "fact": 

234 return False 

235 tags = set(item.tags or ()) 

236 if item.id not in state.promoted and ("private" in tags or ("captured" in tags and not share_captured)): 

237 return False 

238 return bool(PUSH_ID_RE.fullmatch(item.id)) 

239 

240 

241def _record(item: Any) -> dict[str, Any]: 

242 from astrocyte._sync import content_hash 

243 

244 metadata = { 

245 k: v for k, v in (item.metadata or {}).items() 

246 if (not k.startswith("_") or k in _PUSHED_SYSTEM_KEYS) and (v is None or isinstance(v, (str, int, float, bool))) 

247 } 

248 return { 

249 "id": item.id, 

250 "text": item.text, 

251 "occurred_at": item.occurred_at.isoformat() if item.occurred_at else None, 

252 "tags": list(item.tags) if item.tags else None, 

253 "fact_type": item.fact_type, 

254 "metadata": metadata or None, 

255 "content_hash": content_hash(item.text), 

256 } 

257 

258 

259async def _all_items(store: Any, bank: str) -> list[Any]: 

260 items, offset = [], 0 

261 while True: 

262 page = await store.list_vectors(bank, offset=offset, limit=500) 

263 items += page 

264 if len(page) < 500: 

265 return items 

266 offset += 500 

267 

268 

269async def pending_push(store: Any, bank: str, state: SyncState, *, share_captured: bool) -> list[Any]: 

270 """Local memories that would go up on the next push, oldest first.""" 

271 items = [i for i in await _all_items(store, bank) 

272 if i.id not in state.pushed and shareable(i, state, share_captured=share_captured)] 

273 epoch = datetime.min.replace(tzinfo=timezone.utc) 

274 return sorted(items, key=lambda i: (i.retained_at or i.occurred_at or epoch, i.id)) 

275 

276 

277# ── the gateway ────────────────────────────────────────────────────────── 

278 

279 

280def _client(url: str, token: str) -> Any: 

281 """An HTTP client for the gateway (tests substitute a transport).""" 

282 import httpx 

283 

284 return httpx.AsyncClient(base_url=url, headers={"Authorization": f"Bearer {token}"}, timeout=30.0) 

285 

286 

287def _bank_path(bank: str, rest: str) -> str: 

288 return f"/v1/banks/{quote(bank, safe='')}/{rest}" 

289 

290 

291async def _call(client: Any, method: str, path: str, **kw: Any) -> Any: 

292 import httpx 

293 

294 try: 

295 response = await client.request(method, path, **kw) 

296 except httpx.HTTPError as e: 

297 raise TeamError(f"can't reach the gateway at {client.base_url}: {e}") from None 

298 if response.status_code == 401: 

299 raise TeamError("the gateway refused the token (401): check it, or ask for a new one") 

300 if response.status_code == 403: 

301 raise TeamError(f"the token has no access to this project's bank (403): {_detail(response)}") 

302 if response.status_code == 423: 

303 raise TeamError(f"the project is under a legal hold on the gateway, so nothing was forgotten: {_detail(response)}") 

304 if response.status_code == 501: 

305 raise TeamError("this gateway's storage doesn't support team sync (501): it needs the Postgres or " 

306 "SQLite store") 

307 if response.status_code >= 400: 

308 raise TeamError(f"the gateway answered {response.status_code}: {_detail(response)}") 

309 return response.json() 

310 

311 

312def _detail(response: Any) -> str: 

313 try: 

314 return str(response.json().get("detail") or response.text)[:300] 

315 except ValueError: 

316 return response.text[:300] 

317 

318 

319async def probe(client: Any, bank: str) -> None: 

320 """Reachable, the token accepted, and ``read`` on the bank.""" 

321 await _call(client, "GET", _bank_path(bank, "changes"), params={"limit": 1}) 

322 

323 

324async def check(bank: str, entry: dict[str, Any]) -> None: 

325 """:func:`probe` a joined project's gateway with its saved token.""" 

326 async with _connect(bank, entry) as client: 

327 await probe(client, bank) 

328 

329 

330@dataclass 

331class SyncReport: 

332 pushed: dict[str, int] = field(default_factory=dict) # status → count 

333 pulled: int = 0 

334 erased: int = 0 

335 skipped: int = 0 # duplicates of a local memory, or refused by the local policy 

336 

337 def summary(self) -> str: 

338 sent = sum(self.pushed.values()) 

339 parts = [f"pushed {sent}" + (f" ({_counts(self.pushed)})" if sent else "")] 

340 parts.append(f"pulled {self.pulled}") 

341 if self.erased: 

342 parts.append(f"erased {self.erased} forgotten by the team") 

343 if self.skipped: 

344 parts.append(f"skipped {self.skipped} already here") 

345 return ", ".join(parts) 

346 

347 

348def _counts(by_status: dict[str, int]) -> str: 

349 return ", ".join(f"{n} {status}" for status, n in sorted(by_status.items()) if n) 

350 

351 

352async def push(client: Any, store: Any, bank: str, state: SyncState, *, share_captured: bool, 

353 report: SyncReport) -> None: 

354 items = await pending_push(store, bank, state, share_captured=share_captured) 

355 for start in range(0, len(items), PUSH_BATCH): 

356 batch = items[start:start + PUSH_BATCH] 

357 reply = await _call(client, "POST", _bank_path(bank, "sync/push"), 

358 json={"records": [_record(i) for i in batch]}) 

359 for result in reply.get("results") or []: 

360 status = result.get("status") 

361 if status == "duplicate": 

362 state.pushed[result["id"]] = f"duplicate:{result.get('duplicate_of')}" 

363 elif status == "rejected": 

364 state.pushed[result["id"]] = f"rejected:{result.get('reason')}" 

365 else: 

366 state.pushed[result["id"]] = str(status) 

367 report.pushed[status] = report.pushed.get(status, 0) + 1 

368 state.save(bank) # each acknowledged batch, so a failure later never re-sends it 

369 

370 

371async def pull(client: Any, pipeline: Any, brain: Any, bank: str, state: SyncState, report: SyncReport, 

372 *, apply: bool = True) -> int: 

373 """Apply the changes feed from the saved cursor; returns how many changes 

374 it read. ``apply=False`` only counts them (and moves no cursor).""" 

375 from astrocyte.types import SyncPushRecord 

376 

377 cursor, seen = state.cursor, 0 

378 pulled = set(state.pulled) 

379 while True: 

380 params: dict[str, Any] = {"limit": PULL_PAGE} 

381 if cursor: 

382 params["cursor"] = cursor 

383 page = await _call(client, "GET", _bank_path(bank, "changes"), params=params) 

384 changes = page.get("changes") or [] 

385 seen += len(changes) 

386 if apply and changes: 

387 erase: list[str] = [] 

388 incoming: list[SyncPushRecord] = [] 

389 for c in changes: 

390 cid = c.get("id") 

391 if not isinstance(cid, str): 

392 continue 

393 if c.get("deleted"): 

394 if cid in state.withheld: 

395 continue # our own memory, taken off the team with `unshare`: it stays here 

396 erase.append(cid) 

397 continue 

398 own = cid in state.pushed and not state.pushed[cid].startswith(("duplicate:", "rejected:")) 

399 if own or cid in state.suppressed or (c.get("memory_layer") or "fact") != "fact" or not c.get("text"): 

400 continue # ours already; server-made observations and models aren't mirrored 

401 incoming.append(SyncPushRecord( 

402 id=cid, text=c["text"], occurred_at=_parse_time(c.get("occurred_at")), 

403 tags=c.get("tags") or None, fact_type=c.get("fact_type"), 

404 metadata={k: v for k, v in (c.get("metadata") or {}).items() 

405 if v is None or isinstance(v, (str, int, float, bool))} or None, 

406 )) 

407 if incoming: 

408 for result in await brain.push_records(bank, incoming): 

409 if result.status in ("stored", "unchanged"): 

410 if result.id not in pulled: 

411 pulled.add(result.id) 

412 state.pulled.append(result.id) 

413 report.pulled += result.status == "stored" 

414 else: 

415 report.skipped += 1 

416 if erase: 

417 report.erased += await _erase_local(pipeline, brain, bank, erase) 

418 for cid in erase: 

419 if cid in pulled: 

420 pulled.discard(cid) 

421 state.pulled.remove(cid) 

422 if cid in state.pushed: 

423 state.pushed[cid] = "forgotten" 

424 cursor = page.get("next_cursor") or cursor 

425 if apply: 

426 state.cursor = cursor 

427 state.save(bank) 

428 if not page.get("has_more"): 

429 return seen 

430 

431 

432async def _erase_local(pipeline: Any, brain: Any, bank: str, ids: list[str]) -> int: 

433 """Forget and purge ``ids`` here, as ``astrocyte memory forget`` does.""" 

434 present = set(await _existing(pipeline.vector_store, bank, ids)) 

435 if not present: 

436 return 0 

437 result = await brain.forget(bank, memory_ids=sorted(present)) 

438 purge = getattr(pipeline.vector_store, "purge", None) 

439 if purge is not None: 

440 await purge(bank, sorted(present)) 

441 return result.deleted_count 

442 

443 

444async def _existing(store: Any, bank: str, ids: list[str]) -> list[str]: 

445 """Which of ``ids`` are live memories in ``bank`` here.""" 

446 lookup = getattr(store, "lookup_ids", None) 

447 if lookup is not None: 

448 return [c.id for c in await lookup(ids) if not c.deleted and c.bank_id == bank] 

449 wanted = set(ids) 

450 return [i.id for i in await _all_items(store, bank) if i.id in wanted] 

451 

452 

453def _parse_time(value: Any) -> datetime | None: 

454 if not isinstance(value, str): 

455 return None 

456 try: 

457 return datetime.fromisoformat(value.replace("Z", "+00:00")) 

458 except ValueError: 

459 return None 

460 

461 

462async def forget_shared(entry: dict[str, Any], bank: str, ids: list[str], pipeline: Any, brain: Any, *, 

463 team_wide: bool) -> int: 

464 """Erase ``ids`` here; with ``team_wide``, first on the gateway too, so 

465 every teammate's next sync erases them (needs ``forget`` on the bank). 

466 Without it, the ids are suppressed so a later pull doesn't bring them 

467 back. Returns how many memories were erased here.""" 

468 with _BankLock(bank): 

469 state = SyncState.load(bank) 

470 shared = [i for i in ids if state.shared(i)] 

471 if team_wide and shared: 

472 async with _connect(bank, entry) as client: 

473 # Erased, not only forgotten: the text leaves the gateway's storage too. 

474 await _call(client, "POST", "/v1/forget", json={"bank_id": bank, "memory_ids": shared, "erase": True}) 

475 erased = await _erase_local(pipeline, brain, bank, ids) 

476 for mid in ids: 

477 if mid in state.pulled: 

478 state.pulled.remove(mid) 

479 if team_wide and mid in state.pushed: 

480 state.pushed[mid] = "forgotten" 

481 if not team_wide and mid in shared and mid not in state.suppressed: 

482 state.suppressed.append(mid) 

483 state.save(bank) 

484 return erased 

485 

486 

487def _same_retain(items: list[Any], ids: list[str]) -> list[str]: 

488 """``ids`` plus every chunk retained with them: a long turn is several 

489 rows sharing a ``_retain_id``, and sharing half of one makes no sense.""" 

490 chosen = set(ids) 

491 keys = {(i.metadata or {}).get("_retain_id") for i in items if i.id in chosen} - {None} 

492 return [i.id for i in items if i.id in chosen or (i.metadata or {}).get("_retain_id") in keys] 

493 

494 

495async def share(bank: str, ids: list[str], store: Any) -> list[str]: 

496 """Share these memories (and their chunks) at the next push, whatever the 

497 rules say. Returns the ids that will go up; a teammate's are skipped.""" 

498 with _BankLock(bank): 

499 state = SyncState.load(bank) 

500 ids = [i for i in _same_retain(await _all_items(store, bank), ids) if i not in state.pulled] 

501 gone = [i for i in ids if state.pushed.get(i) in ("unshared", "forgotten")] 

502 if gone: 

503 # The gateway keeps a forgotten id forgotten: a re-push is rejected. 

504 raise TeamError(f"{gone[0][:8]} was taken off the team's gateway, and a forgotten memory can't be " 

505 "shared again under its id") 

506 for mid in ids: 

507 if mid in state.withheld: 

508 state.withheld.remove(mid) 

509 if mid not in state.promoted: 

510 state.promoted.append(mid) 

511 state.save(bank) 

512 return ids 

513 

514 

515async def unshare(entry: dict[str, Any] | None, bank: str, ids: list[str], store: Any) -> tuple[list[str], int]: 

516 """Keep these memories (and their chunks) on this machine. One already 

517 on the gateway is forgotten there (needs ``forget``), so every teammate's 

518 copy goes at their next sync; yours stays. Returns (ids kept here, how 

519 many were taken off the gateway). A teammate's memory can't be unshared.""" 

520 with _BankLock(bank): 

521 state = SyncState.load(bank) 

522 ids = _same_retain(await _all_items(store, bank), ids) 

523 theirs = [i for i in ids if i in state.pulled] 

524 if theirs: 

525 raise TeamError(f"{theirs[0][:8]} is a teammate's memory: hide it here with " 

526 "`astrocyte memory forget <id> --local`") 

527 on_gateway = [i for i in ids if state.pushed.get(i) in ("stored", "unchanged")] 

528 if on_gateway: 

529 if entry is None: 

530 raise TeamError("these memories are on the team's gateway, but this project isn't joined any more") 

531 async with _connect(bank, entry) as client: 

532 await _call(client, "POST", "/v1/forget", 

533 json={"bank_id": bank, "memory_ids": on_gateway, "erase": True}) 

534 for mid in ids: 

535 if mid in state.promoted: 

536 state.promoted.remove(mid) 

537 if mid not in state.withheld: 

538 state.withheld.append(mid) 

539 if mid in on_gateway: 

540 state.pushed[mid] = "unshared" 

541 state.save(bank) 

542 return ids, len(on_gateway) 

543 

544 

545# ── commands ───────────────────────────────────────────────────────────── 

546 

547 

548def _bank(args: Namespace) -> str: 

549 from .project import project_bank 

550 

551 return args.bank or project_bank(str(Path(args.project or os.getcwd()).expanduser().resolve())) 

552 

553 

554def _membership(cfg: Path, bank: str) -> dict[str, Any]: 

555 entry = load_memberships(cfg).get(bank) 

556 if entry is None: 

557 raise TeamError(f"{bank} isn't shared with a team. Join first: astrocyte team join <gateway-url> --token …") 

558 return entry 

559 

560 

561def _connect(bank: str, entry: dict[str, Any]) -> Any: 

562 token = token_for(bank, entry) 

563 if not token: 

564 raise TeamError(f"no token for {entry['url']}: run astrocyte team join again") 

565 return _client(entry["url"], token) 

566 

567 

568def _preview(items: list[Any]) -> None: 

569 from .agentd import render_memory 

570 

571 for item in items[:PREVIEW_SAMPLE]: 

572 print(" " + render_memory(item.text, None).removeprefix("- ")[:110]) 

573 if len(items) > PREVIEW_SAMPLE: 

574 print(f" … and {len(items) - PREVIEW_SAMPLE} more") 

575 

576 

577async def sync_bank(entry: dict[str, Any], bank: str, pipeline: Any, brain: Any) -> SyncReport: 

578 """Push, then pull, one joined bank; the state records the outcome either 

579 way. Raises :class:`SyncBusy` if another process is syncing it.""" 

580 with _BankLock(bank): 

581 state = SyncState.load(bank) 

582 report = SyncReport() 

583 try: 

584 async with _connect(bank, entry) as client: 

585 await push(client, pipeline.vector_store, bank, state, 

586 share_captured=bool(entry.get("share_captured")), report=report) 

587 await pull(client, pipeline, brain, bank, state, report) 

588 except TeamError as e: 

589 state = SyncState.load(bank) # keep what push and pull saved as they went 

590 state.last_error = str(e) 

591 state.save(bank) 

592 raise 

593 state.last_sync = datetime.now(timezone.utc).isoformat(timespec="seconds") 

594 state.last_error = None 

595 state.save(bank) 

596 return report 

597 

598 

599async def _sync(entry: dict[str, Any], bank: str, pipeline: Any, brain: Any, *, dry_run: bool) -> int: 

600 if dry_run: 

601 state = SyncState.load(bank) 

602 async with _connect(bank, entry) as client: 

603 items = await pending_push(pipeline.vector_store, bank, state, 

604 share_captured=bool(entry.get("share_captured"))) 

605 print(f"Would push {len(items)} memories to {entry['url']}:") 

606 _preview(items) 

607 waiting = await pull(client, pipeline, brain, bank, state, SyncReport(), apply=False) 

608 print(f"Would pull {waiting} changes from the team.") 

609 return 0 

610 report = await sync_bank(entry, bank, pipeline, brain) 

611 print(f"Synced {bank}: {report.summary()}.") 

612 return 0 

613 

614 

615async def _join(args: Namespace, cfg: Path, pipeline: Any, brain: Any) -> int: 

616 bank = _bank(args) 

617 url = args.url.rstrip("/") 

618 if not url.startswith(("http://", "https://")): 

619 raise TeamError(f"{args.url}: give the gateway's URL, e.g. https://memory.example.com") 

620 token = args.token or os.environ.get("ASTROCYTE_TEAM_TOKEN") 

621 if not token: 

622 raise TeamError("give the token the gateway's admin created for you: --token … (or ASTROCYTE_TEAM_TOKEN)") 

623 async with _client(url, token) as client: 

624 await probe(client, bank) 

625 entry = {"url": url, "share_captured": bool(args.share_captured), 

626 "joined_at": datetime.now(timezone.utc).isoformat(timespec="seconds")} 

627 state = SyncState.load(bank) 

628 items = await pending_push(pipeline.vector_store, bank, state, share_captured=entry["share_captured"]) 

629 already = load_memberships(cfg).get(bank) 

630 print(f"Updating {bank}'s team settings ({url})." if already else f"Joining {bank} at {url}.") 

631 if items: 

632 print(f"This shares {len(items)} memories from this machine with everyone who can read the bank" 

633 + (" (captured conversation included)" if entry["share_captured"] else 

634 " (saved and imported ones; captured conversation stays here)") + ":") 

635 _preview(items) 

636 if not args.yes: 

637 if not sys.stdin.isatty(): 

638 print("Re-run with --yes to share them.", file=sys.stderr) 

639 return 1 

640 if input("Share them? [y/N] ").strip().lower() not in ("y", "yes"): 

641 print("Not joined; nothing left this machine.") 

642 return 1 

643 else: 

644 print("Nothing on this machine to share yet; teammates' memories will be pulled.") 

645 projects = load_memberships(cfg) 

646 where = store_token(bank, url, token, entry) 

647 projects[bank] = entry 

648 save_memberships(cfg, projects)