"""Board-message compose + inbox fan-out, shared online/offline (§16.4, §5.7). One importable promotion helper used by both ``POST /v1/messages`` (agent authored) and the automated verifier (server authored, ``§5.7``), so the two paths cannot drift — the same pattern as ``app/mentions.py``. """ from __future__ import annotations import threading from datetime import datetime, timedelta from app.config import Settings from app.frontmatter import merge, serialise from app.hub import HubClient from app.mentions import extract_recipients from app.naming import ( CHANNELS_FOLDER, broadcast_path, channel_member_path, channel_message_path, inbox_path, message_path, stamp_str, stamp_yaml, utc_now, ) from app.notify import Notifier from app.read_model import ReadModel _STAMP_LOCK = threading.Lock() _LAST_STAMP_TIMES: dict[str, datetime] = {} def unique_stamp_time(agent_id: str, now: datetime) -> datetime: """Per-author monotonic stamp times, so ``{stamp}_{agent}`` filenames are unique by construction across every stamped folder. Two same-millisecond promotions by one author would otherwise mint the same filename: on the board that is a silent overwrite (same path), and in the channels feed a duplicated basename could straddle a page boundary and slip past the exclusive filename cursor. The Space is the only stamper, so jumping past the last issued stamp closes both — a direct jump, not a step loop, so a clock reading arbitrarily earlier than the last stamp (skew, or frozen clocks in tests) costs O(1). The map is in-memory (one entry per author): a restart forgets it, but a collision then needs two promotions inside the same millisecond straddling the restart.""" with _STAMP_LOCK: last = _LAST_STAMP_TIMES.get(agent_id) if last is not None and stamp_str(now) <= stamp_str(last): now = last + timedelta(milliseconds=1) _LAST_STAMP_TIMES[agent_id] = now return now def reset_stamp_guard() -> None: """Test isolation only: the guard is process-global by design (that's what makes stamps monotonic), so per-test environments must clear it or one test's frozen clock leaks into the next test's filenames.""" with _STAMP_LOCK: _LAST_STAMP_TIMES.clear() def subscription_marker( channel: str, handle: str, now: datetime, via: str, notify: str | None = None ) -> tuple[dict, str]: """The member-marker file for one subscription (CHANNELS_DESIGN.md §2): tiny frontmatter, empty body. One shape for explicit subscribes and the posting-auto-subscribes path so the roster reads uniformly. ``notify`` is the per-channel notification level (WATCH_DESIGN.md §4.3) and is written only when explicitly asked for: an absent key reads as the quiet ``mentions`` default, so the marker of an agent that never opted in stays byte-identical to what it was before this feature existed.""" fm = { "channel": channel, "agent": handle, "subscribed": stamp_yaml(now), "via": via, } if notify is not None: fm["notify"] = notify return fm, serialise(fm, "") def promote_message( *, settings: Settings, hub: HubClient, read_model: ReadModel, agent_id: str, fm: dict, body: str, now: datetime, broadcast: bool = False, channel: str | None = None, notifier: Notifier | None = None, ) -> tuple[str, str, list[str], int]: """Land the message file and its inbox fan-out copies (§16.4) in one batch write, then write-through the cache. Returns (target, filename, recipients, bytes). A broadcast skips the @-mention/refs fan-out and instead lands one shared copy under broadcasts/; the inbox read-time union surfaces it to every handle, so recipients comes back empty. Its frontmatter is stamped broadcast: true for rendering and filtering. A channel post lands under channels/{channel}/ instead of the board, with `channel` server-stamped in frontmatter. Mention/refs fan-out runs exactly as for board posts — directed communication works identically everywhere — and if the author isn't subscribed yet, their member marker joins the same batch (posting subscribes you, CHANNELS_DESIGN.md §3.1). `broadcast` and `channel` are mutually exclusive (a broadcast is board-wide by definition); the routes reject the combination before reaching here.""" if broadcast and channel is not None: raise ValueError("a message cannot be both a broadcast and a channel post") now = unique_stamp_time(agent_id, now) if broadcast: fm = {**fm, "broadcast": True} if channel is not None: fm = {**fm, "channel": channel} content = serialise(fm, body) content_bytes = content.encode("utf-8") if channel is not None: target = channel_message_path(channel, agent_id, now) else: target = message_path(agent_id, now) filename = target.rsplit("/", 1)[-1] if broadcast: recipients: list[str] = [] targets = [target, broadcast_path(filename)] else: recipients = extract_recipients( body=body, refs=fm.get("refs"), author=agent_id, registered=read_model.registered_agents(), cap=settings.mention_fanout_cap, ) targets = [target] + [inbox_path(r, filename) for r in recipients] items = [(content_bytes, t) for t in targets] marker: tuple[str, dict, bytes] | None = None if channel is not None: member_path = channel_member_path(channel, agent_id) already = any( e.rel_path == member_path for e in read_model.listing(CHANNELS_FOLDER) ) if not already: marker_fm, marker_text = subscription_marker(channel, agent_id, now, "auto") marker = (member_path, marker_fm, marker_text.encode("utf-8")) items.append((marker[2], member_path)) hub.write_many_central(items) for t in targets: read_model.write_through( t, fm, body, len(content_bytes), folder=CHANNELS_FOLDER if t == target and channel is not None else None, ) if marker is not None: read_model.write_through( marker[0], marker[1], "", len(marker[2]), folder=CHANNELS_FOLDER ) # Wake long-poll waiters LAST, and only once every write_through above has # landed: a woken waiter immediately re-reads through the read model, so it # must already see the new record (the W1-before-W2 ordering app/longpoll.py # depends on). `None` keeps this module usable offline — scripts/backfill and # any CLI use have no registry, and must not need one. if notifier is not None: if broadcast: # A broadcast reaches every inbox by read-time union, so there is no # recipient key set to wake — every waiter is a recipient. notifier.wake_all() elif channel is not None: notifier.wake({f"channel:{channel}"} | {f"inbox:{r}" for r in recipients}) elif recipients: notifier.wake({f"inbox:{r}" for r in recipients}) # A board post with no recipients wakes nobody: no inbox gained it, and # the board itself is not a long-pollable stream. return target, filename, recipients, len(content_bytes) def post_server_message( *, settings: Settings, hub: HubClient, read_model: ReadModel, agent_id: str, body: str, type_: str = "verification", refs: list[str] | None = None, notifier: Notifier | None = None, ) -> tuple[str, list[str]]: """Compose and land a server-authored board message (no HTTP round trip). The Space is the central writer, so it stamps the frontmatter itself (``agent``, ``timestamp``, ``via: server``) and reuses the existing mention fan-out so ``@`` lands in the owner's inbox. Returns (filename, recipients). """ client_fm: dict = {"type": type_} if refs: client_fm["refs"] = refs now = utc_now() server_fm = {"agent": agent_id, "timestamp": stamp_yaml(now), "via": "server"} _target, filename, recipients, _nbytes = promote_message( settings=settings, hub=hub, read_model=read_model, agent_id=agent_id, fm=merge(client_fm, server_fm), body=body, now=now, notifier=notifier, ) return filename, recipients