Coverage for astrocyte/harness/agentd.py: 93%

1"""``astrocyte agentd`` — warm memory for agent hooks. 

2 

3Why a daemon: a hook is a fresh process per event. Measured on an Apple 

4Silicon laptop, the first recall in a fresh process costs ~1.1 s (mostly 

5importing fastembed), against 8 ms in a warm one — and ``UserPromptSubmit`` 

6blocks the user's prompt until the hook returns. So hooks talk to this 

7per-user process and never load a model themselves. 

8 

9Transport: a Unix domain socket in the state directory (0600: filesystem 

10permissions are the authentication). Where there is none (Windows; or forced 

11with ``ASTROCYTE_AGENTD_TRANSPORT=tcp``), TCP on 127.0.0.1 at a port the OS 

12picks, published with a random token in ``agentd.json`` in the state 

13directory; a request without that token gets no reply. 

14 

15Lifecycle: started on demand by the hooks, one instance per user (a lock file), 

16exits after ``ASTROCYTE_AGENTD_IDLE`` seconds without a request (default 

171800) or as soon as its config file changes or the package is upgraded, so the 

18next hook restarts it on the new config and code. Captures arrive through a durable on-disk spool, so a daemon 

19that is down or restarting loses nothing. 

20 

21Relevance gating: recall's fused ``score`` is a ranking value, not a 

22similarity — an unrelated prompt ("weather in Paris") outranked related ones 

23in testing. Injection is therefore gated on the cosine between the prompt and 

24each candidate, computed here with the warm model. Calibrated on 

25bge-small-en-v1.5: relevant pairs scored ≥ 0.661 (long captured turns) and 

26≥ 0.706 (short facts); unrelated prompts scored ≤ 0.609. Defaults sit in that 

27gap; ``ASTROCYTE_INJECT_MIN_SIMILARITY`` overrides. 

28""" 

29 

30from __future__ import annotations 

31 

32import argparse 

33import contextlib 

34import hmac 

35import json 

36import logging 

37import logging.handlers 

38import math 

39import os 

40import re 

41import secrets 

42import socket 

43import subprocess 

44import sys 

45import time 

46import uuid 

47from datetime import datetime, timezone 

48from pathlib import Path 

49from typing import TYPE_CHECKING, Any 

50 

51if TYPE_CHECKING: 

52 import asyncio 

53 

54from .paths import agentd_endpoint, agentd_socket, config_path, spool_dir, state_dir 

55 

56logger = logging.getLogger("astrocyte.agentd") 

57 

58INJECT_MIN_SIMILARITY = float(os.environ.get("ASTROCYTE_INJECT_MIN_SIMILARITY", "0.65")) 

59# Keep only hits close to the best one: weaker neighbours that clear the 

60# absolute bar were, in calibration, other topics from the same session. 

61INJECT_RELATIVE_WINDOW = 0.08 

62PROMPT_MAX_HITS = 4 

63PROMPT_MAX_CHARS = 1_800 

64BOOT_MAX_ITEMS = 8 

65BOOT_MAX_CHARS = 3_000 

66ITEM_MAX_CHARS = 360 

67# Where you left off: the closing turns of the project's previous session. 

68# The last answer gets the most room: that is where conclusions and next steps are. 

69# Opt-in file recall: earlier turns that touched the file the agent just 

70# read or edited (their `files` metadata, recorded at capture). 

71FILE_MAX_TURNS = 3 

72FILE_SCAN = 400 

73RESUME_MAX_TURNS = 3 

74RESUME_LAST_MAX_CHARS = 700 

75RESUME_SCAN = 60 

76SOURCE_LABELS = {"claude-code": "Claude Code", "codex": "Codex", "antigravity": "Antigravity", 

77 "copilot": "Copilot CLI"} 

78DRAIN_INTERVAL_SECONDS = 20.0 

79#: Background team sync (`astrocyte team join`): every joined project, this often. 

80TEAM_SYNC_SECONDS = float(os.environ.get("ASTROCYTE_TEAM_SYNC_SECONDS", "300")) 

81IDLE_EXIT_SECONDS = float(os.environ.get("ASTROCYTE_AGENTD_IDLE", "1800")) 

82 

83# The conversation pipeline stores turns with single newlines between roles, 

84# whatever separator capture wrote, so match any run of whitespace. 

85_TURN = re.compile(r"^\*\*user\*\*:\s*(?P<q>.*?)\s*\*\*assistant\*\*:\s*(?P<a>.*)$", re.DOTALL) 

86 

87 

88TRANSPORT_ENV = "ASTROCYTE_AGENTD_TRANSPORT" 

89 

90 

91def transport() -> str: 

92 """``unix`` where there are Unix domain sockets, else ``tcp`` (loopback + 

93 token). ``ASTROCYTE_AGENTD_TRANSPORT=tcp`` forces TCP, so the Windows 

94 path is exercised on every platform.""" 

95 unix = hasattr(socket, "AF_UNIX") and os.name != "nt" 

96 forced = os.environ.get(TRANSPORT_ENV, "").strip().lower() 

97 return "tcp" if forced == "tcp" or not unix else "unix" 

98 

99 

100def supported() -> bool: 

101 """Can hooks reach a daemon here? Every platform has a transport; kept so 

102 callers can still ask (and tests can say no).""" 

103 return True 

104 

105 

106# ── rendering ──────────────────────────────────────────────────────────── 

107 

108 

109def _clip(text: str, limit: int) -> str: 

110 text = " ".join(text.split()) 

111 return text if len(text) <= limit else text[: limit - 1].rstrip() + "…" 

112 

113 

114def who_saved(metadata: Any) -> str | None: 

115 """A teammate's name from the ``_actor`` the gateway stamped: ``user:alice`` 

116 reads as ``alice``; other principals (``service:ci``) as they are.""" 

117 actor = (metadata or {}).get("_actor") 

118 if not isinstance(actor, str) or not actor: 

119 return None 

120 kind, _, name = actor.partition(":") 

121 return name if kind == "user" and name else actor 

122 

123 

124def _line(body: str, when: datetime | None, who: str | None) -> str: 

125 return "- " + (f"[{when.date().isoformat()}] " if when else "") + (f"({who}) " if who else "") + body 

126 

127 

128def render_memory(text: str, when: datetime | None, *, limit: int = ITEM_MAX_CHARS, who: str | None = None) -> str: 

129 """One line per memory. Captured turns become "Q … → A …"; the answer 

130 gets most of the room because that is where conclusions live. ``who`` 

131 names the teammate a memory came from; your own carry no label.""" 

132 m = _TURN.match(text.strip()) 

133 if m: 

134 q = _clip(m["q"], 110) 

135 body = f"Q: {q} → A: {_clip(m['a'], limit - len(q) - 12)}" 

136 else: 

137 body = _clip(text, limit) 

138 return _line(body, when, who) 

139 

140 

141def _clip_left(text: str, limit: int) -> str: 

142 """The end of ``text``: where a long answer's conclusion is.""" 

143 text = " ".join(text.split()) 

144 if len(text) <= limit: 

145 return text 

146 tail = text[-(limit - 1):] 

147 return "…" + (tail.split(" ", 1)[1] if " " in tail[:40] else tail) 

148 

149 

150def _join_overlapping(a: str, b: str) -> str: 

151 """``a`` then ``b``, written once where the chunker made them overlap.""" 

152 for k in range(min(len(a), len(b)), 15, -1): 

153 if a.endswith(b[:k]): 

154 return a + b[k:] 

155 return f"{a} {b}" if a else b 

156 

157 

158def _groups(items: list[Any]) -> list[list[Any]]: 

159 """Memories regrouped into what was retained together, newest first. 

160 

161 A long captured turn is stored as several chunks (the question, then 

162 overlapping pieces of the answer) that share one ``_retain_id`` and come 

163 back in no useful order; ``_chunk_index`` gives the order they were written.""" 

164 groups: dict[str, list[Any]] = {} 

165 for item in items: 

166 meta = item.metadata or {} 

167 # _retain_id since 0.18; _created_at alone for memories stored before 

168 # (it can merge retains made within one clock tick). 

169 key = meta.get("_retain_id") or meta.get("_created_at") or item.id 

170 groups.setdefault(str(key), []).append(item) 

171 out = [] 

172 for group in groups.values(): 

173 # _chunk_index where stored (since 0.18); retained_at for older chunks. 

174 group.sort(key=lambda i: ((i.metadata or {}).get("_chunk_index", 0), i.retained_at is None, 

175 i.retained_at or 0)) 

176 out.append(group) 

177 return out 

178 

179 

180def render_group(group: list[Any], when: datetime | None, *, limit: int = ITEM_MAX_CHARS, 

181 who: str | None = None) -> str: 

182 """One line for memories retained together. A chunked turn shows its 

183 question and the end of its answer; other text, its beginning.""" 

184 if len(group) == 1: 

185 return render_memory(group[0].text, when, limit=limit, who=who) 

186 question = next((i.text for i in group if i.text.lstrip().startswith("**user**")), None) 

187 if question is None: 

188 return render_memory(group[0].text, when, limit=limit, who=who) 

189 m = _TURN.match(question.strip()) 

190 q = _clip(m["q"] if m else question.split(":", 1)[-1], 110) 

191 answer = "" 

192 for item in group: 

193 if item.text.lstrip().startswith("**user**"): 

194 answer = m["a"] if m else "" 

195 else: 

196 answer = _join_overlapping(answer, item.text.strip()) 

197 answer = re.sub(r"^\*\*assistant\*\*:\s*", "", answer.strip()) 

198 body = f"Q: {q} → A: {_clip_left(answer, limit - len(q) - 12)}" 

199 return _line(body, when, who) 

200 

201 

202def _budget(lines: list[str], limit: int) -> list[str]: 

203 out, used = [], 0 

204 for line in lines: 

205 if used + len(line) + 1 > limit: 

206 break 

207 out.append(line) 

208 used += len(line) + 1 

209 return out 

210 

211 

212def _ago(when: datetime, now: datetime | None = None) -> str: 

213 if when.tzinfo is None: 

214 when = when.replace(tzinfo=timezone.utc) 

215 minutes = int(((now or datetime.now(timezone.utc)) - when).total_seconds() // 60) 

216 if minutes < 2: 

217 return "just now" 

218 if minutes < 60: 

219 return f"{minutes} minutes ago" 

220 if minutes < 48 * 60: 

221 hours = minutes // 60 

222 return f"{hours} hour{'s' if hours != 1 else ''} ago" 

223 return f"{minutes // (24 * 60)} days ago" 

224 

225 

226def _cosine(a: list[float], b: list[float]) -> float: 

227 dot = sum(x * y for x, y in zip(a, b)) 

228 na = math.sqrt(sum(x * x for x in a)) 

229 nb = math.sqrt(sum(y * y for y in b)) 

230 return dot / (na * nb) if na and nb else 0.0 

231 

232 

233# ── daemon ─────────────────────────────────────────────────────────────── 

234 

235 

236def installed_version() -> str | None: 

237 """The astrocyte version installed now, read from disk: an upgrade 

238 replaces it under a running daemon, whose code stays the old one.""" 

239 import importlib 

240 import importlib.metadata 

241 

242 importlib.invalidate_caches() 

243 try: 

244 return importlib.metadata.version("astrocyte") 

245 except importlib.metadata.PackageNotFoundError: 

246 return None 

247 

248 

249class AgentDaemon: 

250 def __init__(self, cfg_path: Path) -> None: 

251 from .memories import open_local 

252 

253 self.cfg_path = cfg_path 

254 self.cfg_mtime = cfg_path.stat().st_mtime 

255 self.version = installed_version() 

256 # No recall-time query expansion (an LLM call of 5–9 s via a CLI 

257 # provider; a hook on the prompt path cannot afford it) and no 

258 # observation consolidation (an LLM call per captured turn in the 

259 # background — the user's subscription, unasked; a capture burst 

260 # queued 32 of them). 

261 self.pipeline, self.brain = open_local(cfg_path) 

262 self.injected: dict[str, set[str]] = {} 

263 self.files_offered: dict[str, set[str]] = {} # session → paths already answered 

264 self.token: str | None = None # set when serving over TCP 

265 self.last_activity = time.monotonic() 

266 self.next_team_sync = 0.0 # due at the first housekeeping pass 

267 self._sync_brain: Any = None 

268 self._teammates: dict[str, tuple[float, set[str]]] = {} # bank → (state mtime, pulled ids) 

269 import asyncio # daemon side only: hook processes import this module too 

270 

271 self._drain_lock = asyncio.Lock() 

272 self._stop = asyncio.Event() 

273 

274 async def warm(self) -> None: 

275 """Load the embedding model before accepting requests, so the first 

276 prompt of a session is as fast as the rest.""" 

277 await self.pipeline.llm_provider.embed(["warm"]) 

278 

279 # ── ops ─────────────────────────────────────────────────────────── 

280 

281 async def op_ping(self, _: dict) -> dict: 

282 return {"ok": True, "pid": os.getpid(), "config": str(self.cfg_path)} 

283 

284 def teammates(self, bank: str) -> set[str]: 

285 """Ids of memories pulled from teammates (re-read when sync changes them).""" 

286 from .team import SyncState 

287 

288 try: 

289 mtime = SyncState.path(bank).stat().st_mtime 

290 except OSError: 

291 return set() 

292 cached = self._teammates.get(bank) 

293 if cached is None or cached[0] != mtime: 

294 cached = (mtime, set(SyncState.load(bank).pulled)) 

295 self._teammates[bank] = cached 

296 return cached[1] 

297 

298 def _who(self, group: list[Any], teammates: set[str]) -> str | None: 

299 return who_saved(group[0].metadata) if group[0].id in teammates else None 

300 

301 async def _previous_session(self, bank: str, session: str) -> list[list[Any]]: 

302 """The closing captured turns (oldest first) of the newest session in 

303 ``bank`` other than ``session``. Sessions running side by side 

304 interleave, so other sessions' turns are skipped, not a stop.""" 

305 from astrocyte.types import VectorFilters 

306 

307 recent = getattr(self.pipeline.vector_store, "list_recent_vectors", None) 

308 if recent is None: 

309 return [] 

310 previous, turns = None, [] 

311 teammates = self.teammates(bank) # where *you* left off: a teammate's shared turns aren't it 

312 items = await recent(bank, limit=RESUME_SCAN, filters=VectorFilters(tags=["captured"])) 

313 for group in _groups(items): 

314 sid = (group[0].metadata or {}).get("session_id") or "" 

315 if not sid or sid == session or (previous is not None and sid != previous): 

316 continue 

317 if any(i.id in teammates for i in group): 

318 continue 

319 previous = sid 

320 turns.append(group) 

321 if len(turns) == RESUME_MAX_TURNS: 

322 break 

323 return turns[::-1] 

324 

325 async def op_boot(self, req: dict) -> dict: 

326 bank, session = req["bank"], req.get("session_id") or "" 

327 if req.get("source") in ("clear", "compact"): 

328 self.injected.pop(session, None) # earlier injections left the context 

329 self.files_offered.pop(session, None) 

330 seen = self.injected.setdefault(session, set()) 

331 sections: list[str] = [] 

332 used = 0 

333 resume = await self._previous_session(bank, session) 

334 if resume: 

335 last = resume[-1][-1] 

336 when = last.occurred_at or last.retained_at 

337 agent = SOURCE_LABELS.get(str((last.metadata or {}).get("source") or ""), "an agent") 

338 lines = [render_group(g, None, limit=RESUME_LAST_MAX_CHARS if g is resume[-1] else ITEM_MAX_CHARS) 

339 for g in resume] 

340 sections.append(f"Where you left off: the previous session here ({agent}" 

341 + (f", {_ago(when)}" if when else "") + ") ended with:\n" + "\n".join(lines)) 

342 used = len(sections[0]) 

343 shown = {item.id for g in resume for item in g} 

344 seen |= shown 

345 recent_groups: list[list[Any]] = [] 

346 recent = getattr(self.pipeline.vector_store, "list_recent_vectors", None) 

347 if recent is not None: 

348 items = await recent(bank, limit=(BOOT_MAX_ITEMS + RESUME_MAX_TURNS) * 4) 

349 recent_groups = [g for g in _groups(items) if not shown & {i.id for i in g}][:BOOT_MAX_ITEMS] # not twice 

350 teammates = self.teammates(bank) 

351 body = _budget([render_group(g, g[0].occurred_at or g[0].retained_at, who=self._who(g, teammates)) 

352 for g in recent_groups], BOOT_MAX_CHARS - used) 

353 seen.update(i.id for g in recent_groups[: len(body)] for i in g) # only what was shown counts as injected 

354 header = ( 

355 f"Astrocyte memory is active for this project (bank `{bank}`). Relevant memories from " 

356 "earlier sessions are added to your context automatically; save durable decisions, " 

357 "conventions and preferences with the memory_retain tool." 

358 ) 

359 if not body and not sections: 

360 return {"context": header + " No memories for this project yet."} 

361 if body: 

362 sections.append("Most recent memories:\n" + "\n".join(body)) 

363 return {"context": header + "\n\n" + "\n\n".join(sections)} 

364 

365 async def op_recall(self, req: dict) -> dict: 

366 bank, session, prompt = req["bank"], req.get("session_id") or "", req["prompt"] 

367 result = await self.brain.recall(prompt, bank_id=bank, max_results=8) 

368 hits = [h for h in result.hits if h.text] 

369 if not hits: 

370 return {"context": ""} 

371 vectors = await self.pipeline.llm_provider.embed([prompt] + [h.text for h in hits]) 

372 scored = sorted(((_cosine(vectors[0], v), h) for v, h in zip(vectors[1:], hits)), key=lambda p: -p[0]) 

373 best = scored[0][0] 

374 seen = self.injected.setdefault(session, set()) 

375 teammates = self.teammates(bank) 

376 lines = [] 

377 for sim, hit in scored: 

378 if sim < INJECT_MIN_SIMILARITY or sim < best - INJECT_RELATIVE_WINDOW: 

379 break 

380 key = hit.memory_id or hit.text 

381 if key in seen: 

382 continue 

383 who = who_saved(hit.metadata) if hit.memory_id in teammates else None 

384 lines.append(render_memory(hit.text, hit.occurred_at or hit.retained_at, who=who)) 

385 seen.add(key) 

386 if len(lines) == PROMPT_MAX_HITS: 

387 break 

388 lines = _budget(lines, PROMPT_MAX_CHARS) 

389 if not lines: 

390 return {"context": "", "best_similarity": round(best, 3)} 

391 text = "Possibly relevant memories from earlier sessions (Astrocyte):\n" + "\n".join(lines) 

392 return {"context": text, "best_similarity": round(best, 3)} 

393 

394 async def op_file(self, req: dict) -> dict: 

395 """Earlier captured turns that read or edited ``path``, newest first; 

396 each file once per session, and nothing already in context.""" 

397 from astrocyte.types import VectorFilters 

398 

399 bank, session, path = req["bank"], req.get("session_id") or "", req["path"] 

400 offered = self.files_offered.setdefault(session, set()) 

401 recent = getattr(self.pipeline.vector_store, "list_recent_vectors", None) 

402 if path in offered or recent is None: 

403 return {"context": ""} 

404 offered.add(path) 

405 seen = self.injected.setdefault(session, set()) 

406 groups = [] 

407 for group in _groups(await recent(bank, limit=FILE_SCAN, filters=VectorFilters(tags=["captured"]))): 

408 touched = {f for item in group for f in str((item.metadata or {}).get("files") or "").splitlines()} 

409 if path in touched and not any(item.id in seen for item in group): 

410 groups.append(group) 

411 if len(groups) == FILE_MAX_TURNS: 

412 break 

413 teammates = self.teammates(bank) 

414 body = _budget([render_group(g, g[0].occurred_at or g[0].retained_at, who=self._who(g, teammates)) 

415 for g in groups], PROMPT_MAX_CHARS) 

416 if not body: 

417 return {"context": ""} 

418 seen.update(item.id for g in groups[: len(body)] for item in g) 

419 return {"context": f"Earlier sessions that touched {path} (Astrocyte):\n" + "\n".join(body)} 

420 

421 async def op_capture(self, _: dict) -> dict: 

422 import asyncio 

423 

424 asyncio.get_running_loop().create_task(self.drain()) 

425 return {"ok": True} 

426 

427 # ── capture spool ───────────────────────────────────────────────── 

428 

429 async def drain(self) -> int: 

430 """Retain every spooled turn. Progress is written back after each 

431 turn, so a crash mid-file resumes without duplicating memories.""" 

432 stored = 0 

433 async with self._drain_lock: 

434 for path in sorted(spool_dir().glob("*.json")): 

435 try: 

436 batch = json.loads(path.read_text(encoding="utf-8")) 

437 turns = list(batch["turns"]) 

438 bank = batch["bank"] 

439 except (OSError, ValueError, KeyError, TypeError): 

440 failed = spool_dir() / "failed" 

441 failed.mkdir(exist_ok=True) 

442 with contextlib.suppress(OSError): 

443 path.rename(failed / path.name) 

444 logger.warning("moved unreadable spool file %s aside", path.name) 

445 continue 

446 while turns: 

447 turn = turns[0] 

448 started = turn.get("started_at") 

449 result = await self.brain.retain( 

450 turn["content"], 

451 bank_id=bank, 

452 content_type="conversation", 

453 occurred_at=datetime.fromisoformat(started) if started else None, 

454 metadata={"session_id": batch.get("session_id") or "", "source": batch.get("source") or "", 

455 # Files the turn read or edited, one per line (metadata is flat). 

456 **({"files": "\n".join(turn["files"])} if turn.get("files") else {})}, 

457 tags=["captured"], 

458 source=batch.get("source") or "agent", 

459 ) 

460 if not (getattr(result, "stored", False) or getattr(result, "deduplicated", False)): 

461 logger.error("retain failed for %s: %s", path.name, getattr(result, "error", "?")) 

462 return stored # leave the rest for the next drain 

463 stored += 1 

464 turns.pop(0) 

465 if turns: 

466 _atomic_write(path, {**batch, "turns": turns}) 

467 path.unlink(missing_ok=True) 

468 if stored: 

469 logger.info("captured %d turn(s)", stored) 

470 return stored 

471 

472 # ── server ──────────────────────────────────────────────────────── 

473 

474 async def handle(self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None: 

475 import asyncio 

476 

477 self.last_activity = time.monotonic() 

478 try: 

479 line = await asyncio.wait_for(reader.readline(), timeout=5) 

480 req = json.loads(line or b"{}") 

481 if self.token is not None and not hmac.compare_digest(str(req.pop("token", "")), self.token): 

482 logger.warning("refused a request without the daemon's token") 

483 writer.close() 

484 return 

485 op = getattr(self, f"op_{req.get('op', '')}", None) 

486 reply = await op(req) if op else {"error": f"unknown op {req.get('op')!r}"} 

487 except Exception as e: # noqa: BLE001 — a bad request must not kill the daemon 

488 logger.exception("request failed") 

489 reply = {"error": f"{type(e).__name__}: {e}"} 

490 with contextlib.suppress(Exception): 

491 writer.write((json.dumps(reply) + "\n").encode()) 

492 await writer.drain() 

493 writer.close() 

494 

495 async def team_sync(self) -> None: 

496 """Push and pull every joined project (`astrocyte team join`). A sync 

497 already running elsewhere is skipped; failures are logged and kept in 

498 the project's sync state for `astrocyte team status`.""" 

499 from . import team 

500 

501 memberships = team.load_memberships(self.cfg_path) 

502 if not memberships: 

503 return 

504 if self._sync_brain is None: 

505 # The daemon's own brain refuses a burst of writes as a noisy bank; 

506 # a first pull of a team's memories is exactly such a burst. 

507 from astrocyte import Astrocyte 

508 from astrocyte.config import load_config 

509 

510 from .memories import without_noisy_bank_detection 

511 

512 self._sync_brain = Astrocyte(without_noisy_bank_detection(load_config(str(self.cfg_path)))) 

513 self._sync_brain.set_pipeline(self.pipeline) 

514 for bank, entry in memberships.items(): 

515 try: 

516 report = await team.sync_bank(entry, bank, self.pipeline, self._sync_brain) 

517 logger.info("team sync %s: %s", bank, report.summary()) 

518 except team.SyncBusy: 

519 logger.info("team sync %s: skipped, another sync is running", bank) 

520 except team.TeamError as e: 

521 logger.warning("team sync %s failed: %s", bank, e) 

522 except Exception: # noqa: BLE001 — never let one project's sync stop the daemon 

523 logger.exception("team sync %s failed", bank) 

524 

525 async def housekeeping(self) -> None: 

526 import asyncio 

527 

528 while not self._stop.is_set(): 

529 with contextlib.suppress(asyncio.TimeoutError): 

530 await asyncio.wait_for(self._stop.wait(), timeout=DRAIN_INTERVAL_SECONDS) 

531 if self._stop.is_set(): 

532 break 

533 try: 

534 await self.drain() 

535 except Exception: # noqa: BLE001 

536 logger.exception("drain failed") 

537 if time.monotonic() >= self.next_team_sync: 

538 self.next_team_sync = time.monotonic() + TEAM_SYNC_SECONDS 

539 await self.team_sync() 

540 try: 

541 changed = self.cfg_path.stat().st_mtime != self.cfg_mtime 

542 except OSError: 

543 changed = True 

544 if changed: 

545 logger.info("config changed; exiting so the next hook restarts on it") 

546 self._stop.set() 

547 elif (now := installed_version()) != self.version: 

548 logger.info("astrocyte %s -> %s; exiting so the next hook restarts on it", self.version, now) 

549 self._stop.set() 

550 elif time.monotonic() - self.last_activity > IDLE_EXIT_SECONDS: 

551 logger.info("idle; exiting") 

552 self._stop.set() 

553 

554 async def serve(self, sock_path: Path) -> None: 

555 import asyncio 

556 

557 await self.warm() 

558 if transport() == "tcp": 

559 self.token = secrets.token_urlsafe(32) 

560 server = await asyncio.start_server(self.handle, host="127.0.0.1", port=0) 

561 port = server.sockets[0].getsockname()[1] 

562 endpoint = agentd_endpoint() 

563 _write_private(endpoint, {"transport": "tcp", "port": port, "token": self.token, "pid": os.getpid()}) 

564 logger.info("serving on 127.0.0.1:%d (pid %d)", port, os.getpid()) 

565 cleanup = endpoint 

566 else: 

567 with contextlib.suppress(FileNotFoundError): 

568 sock_path.unlink() # stale: we hold the instance lock 

569 server = await asyncio.start_unix_server(self.handle, path=str(sock_path)) 

570 os.chmod(sock_path, 0o600) 

571 logger.info("serving on %s (pid %d)", sock_path, os.getpid()) 

572 cleanup = sock_path 

573 await self.drain() # anything captured while no daemon was running 

574 async with server: 

575 await self.housekeeping() 

576 with contextlib.suppress(FileNotFoundError): 

577 cleanup.unlink() 

578 

579 

580def _replace(src: Path, dst: Path) -> None: 

581 """``os.replace``, retried briefly: on Windows it fails while another 

582 process (a hook reading state, a virus scanner) has ``dst`` open.""" 

583 for attempt in range(4): 

584 try: 

585 os.replace(src, dst) 

586 return 

587 except PermissionError: 

588 if attempt == 3: 

589 raise 

590 time.sleep(0.05) 

591 

592 

593def _write_private(path: Path, data: dict) -> None: 

594 """Atomically, readable by the owner only (0600 on POSIX; on Windows the 

595 state directory's profile ACLs).""" 

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

597 tmp = path.with_name(f"{path.name}.{os.getpid()}.tmp") 

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

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

600 json.dump(data, fh) 

601 _replace(tmp, path) 

602 

603 

604def _atomic_write(path: Path, data: dict) -> None: 

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

606 tmp.write_text(json.dumps(data), encoding="utf-8") 

607 _replace(tmp, path) 

608 

609 

610def spool_capture(bank: str, session_id: str, source: str, turns: list[dict]) -> Path: 

611 """Durably queue turns for the daemon (atomic rename, 0600).""" 

612 d = spool_dir() 

613 d.mkdir(parents=True, exist_ok=True, mode=0o700) 

614 # Time first so the drain keeps order; the random tail because a coarse 

615 # clock (Windows: ~15 ms) gives two captures in one tick the same name, 

616 # and the second would replace the first. 

617 path = d / f"{time.time_ns()}-{os.getpid()}-{uuid.uuid4().hex[:8]}.json" 

618 tmp = path.with_suffix(".tmp") 

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

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

621 json.dump({"bank": bank, "session_id": session_id, "source": source, "turns": turns}, fh) 

622 os.replace(tmp, path) 

623 return path 

624 

625 

626# ── client ─────────────────────────────────────────────────────────────── 

627 

628 

629def request(op: str, payload: dict[str, Any] | None = None, *, timeout: float = 2.0) -> dict | None: 

630 """One request to the daemon; None if it isn't reachable in time.""" 

631 if not supported(): 

632 return None 

633 message: dict[str, Any] = {"op": op, **(payload or {})} 

634 try: 

635 if transport() == "tcp": 

636 endpoint = json.loads(agentd_endpoint().read_text(encoding="utf-8")) 

637 message["token"] = endpoint["token"] 

638 s = socket.create_connection(("127.0.0.1", int(endpoint["port"])), timeout=timeout) 

639 else: 

640 s = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) 

641 s.settimeout(timeout) 

642 s.connect(str(agentd_socket()))