refactor(lark): read the event-line prefixes from their owner

loopx/extensions/lark/goal_topic_runtime.py

@@ -31,6 +31,7 @@
     build_external_event_response_receipt,
     decide_external_event_ack,
 )
+from .event_collector_runtime import EVENT_DIAGNOSTIC_PREFIX, EVENT_READY_PREFIX
 from .event_inbox import (
     MESSAGE_ID_PATTERN,
     acknowledge_lark_event_inbox,
@@ -120,8 +121,6 @@ def __init__(self, error_code: str, effect_receipt: Mapping[str, Any]) -> None:
     "mentions:(.mentions // .message.mentions // .event.message.mentions // [])}"
 )
 
-_EVENT_READY_PREFIX = "[event] ready "
-_EVENT_DIAGNOSTIC_PREFIX = "[event] "
 _EVENT_EXIT_REASON = re.compile(r"\(reason: (limit|timeout|signal)\)$")
 
 
@@ -750,7 +749,7 @@ def handle_event(line: str) -> None:
             if stop.is_set():
                 break
             stripped = line.strip()
-            if stripped.startswith(_EVENT_READY_PREFIX):
+            if stripped.startswith(EVENT_READY_PREFIX):
                 with result_lock:
                     provider_ready = True
                     if health_sink is not None and not dispatch.failed:
@@ -760,7 +759,7 @@ def handle_event(line: str) -> None:
                 match = _EVENT_EXIT_REASON.search(stripped)
                 exit_reason = match.group(1) if match else None
                 continue
-            if stripped.startswith(_EVENT_DIAGNOSTIC_PREFIX):
+            if stripped.startswith(EVENT_DIAGNOSTIC_PREFIX):
                 continue
             # Batch-shaped provider records are split before scheduling, so an
             # unrelated conversation in a batch can progress independently too.

loopx/extensions/lark/team_plan_confirmation.py

@@ -26,6 +26,7 @@
     read_callback_card_content as _read_callback_card_content,
     update_callback_card as _update_callback_card,
 )
+from .event_collector_runtime import EVENT_DIAGNOSTIC_PREFIX
 from .goal_channel_contracts import (
     binding_for_goal,
     bindings_for_goal,
@@ -64,7 +65,6 @@
 
 
 TEAM_PLAN_CALLBACK_RECEIPT_SCHEMA_VERSION = "lark_team_plan_callback_receipt_v0"
-_EVENT_DIAGNOSTIC_PREFIX = "[event] "
 
 ProcessFactory = Callable[[list[str]], Any]
 ReviewCallbackHandler = Callable[[Mapping[str, Any]], Mapping[str, Any]]
@@ -191,7 +191,7 @@ def consume() -> None:
             if local_stop.is_set() or parent_stop.is_set():
                 return
             stripped = callback_line.strip()
-            if stripped.startswith(_EVENT_DIAGNOSTIC_PREFIX):
+            if stripped.startswith(EVENT_DIAGNOSTIC_PREFIX):
                 continue
             try:
                 callback_event = json.loads(callback_line)

tests/architecture/test_lark_event_prefix_owner.py

@@ -0,0 +1,59 @@
+"""The Lark CLI event-line prefixes have one stating module.
+
+``event_collector_runtime`` already read the provider's stdout lines by their
+prefix. ``goal_topic_runtime`` and ``team_plan_confirmation`` compiled the same
+two bodies locally, so a change to the wire format had three sites to find.
+
+The bodies are stated once, in the owner. A consumer that holds the owner's
+object is fine; a consumer that states the body again fails the census, whether
+it names it ``_EVENT_DIAGNOSTIC_PREFIX`` or anything else.
+"""
+
+from __future__ import annotations
+
+import pathlib
+
+from loopx.extensions.lark import (
+    event_collector_runtime,
+    goal_topic_runtime,
+    team_plan_confirmation,
+)
+
+REPO_ROOT = pathlib.Path(__file__).resolve().parents[2]
+OWNER_PATH = "loopx/extensions/lark/event_collector_runtime.py"
+STATED_LITERALS = ('"[event] ready "', '"[event] "')
+
+
+def _module_sources() -> dict[str, str]:
+    sources: dict[str, str] = {}
+    for path in sorted((REPO_ROOT / "loopx").rglob("*.py")):
+        if "__pycache__" in path.parts:
+            continue
+        sources[path.relative_to(REPO_ROOT).as_posix()] = path.read_text(
+            encoding="utf-8"
+        )
+    return sources
+
+
+def test_only_the_owner_states_each_event_line_prefix() -> None:
+    sources = _module_sources()
+    stated = {
+        literal: sorted(rel for rel, text in sources.items() if literal in text)
+        for literal in STATED_LITERALS
+    }
+    assert stated == {literal: [OWNER_PATH] for literal in STATED_LITERALS}
+
+
+def test_every_consumer_holds_the_owners_object() -> None:
+    assert (
+        goal_topic_runtime.EVENT_READY_PREFIX
+        is event_collector_runtime.EVENT_READY_PREFIX
+    )
+    assert (
+        goal_topic_runtime.EVENT_DIAGNOSTIC_PREFIX
+        is event_collector_runtime.EVENT_DIAGNOSTIC_PREFIX
+    )
+    assert (
+        team_plan_confirmation.EVENT_DIAGNOSTIC_PREFIX
+        is event_collector_runtime.EVENT_DIAGNOSTIC_PREFIX
+    )