fix(ci): make the backend suite importable and green from a cold checkout

The previous commit referenced code that only existed in the working tree,
so CI could not import the modules it changed. Staging by hand cannot see
that: locally every symbol resolves.

- constants, sse_publisher, conversation_utils, forced_alignment,
  vault_verify and timeline/discovery supply symbols the previous commit
  imports (verify_day_episode_ranges, TITLE_NOT_GENERATED,
  publish_sse_event_async, generate_conversation_title,
  generate_short_summary, estimate_words_from_segment_timing). Without
  them collection failed on every test module that reaches the memory
  provider.
- test_streaming_persistence_invariant patched publish_sse_event, which
  the controller no longer exposes at module level.

Two failures CI had and a local run cannot reproduce:

- mask_string's docstring carried \w and \S in a non-raw string. That is a
  SyntaxWarning, which pytest's filterwarnings=error promotes to a
  SyntaxError — but only on an uncached import, so a warm .pyc hides it
  locally and a cold CI checkout does not. Made the docstring raw; it was
  the only such literal in the backend.
- Fixing that unmasked a pre-existing gap it had been aborting before:
  test_memory_setup_wizard loads backends/advanced/init.py, which imports
  chronicle_setup, and tests/unit imports edge/service_manager.py, which
  imports fastapi and uvicorn. None are in the test env. They are supplied
  on the uv command line rather than as path deps, because ../../extras is
  outside the Docker build context.

Verified in a clean worktree with the CI environment rather than the
development one: 1050 backend tests and 120 root tooling tests pass.

Author: AnkushMalaker

.github/workflows/python-tests.yml

@@ -51,11 +51,16 @@ jobs:
           version: "latest"
 
       - name: Run unit tests with coverage
+        # fastapi/uvicorn are the node agent's own runtime deps: tests/unit imports
+        # edge/service_manager.py, which serves the WebUI's service controls. They are
+        # not in setup-requirements.txt, which covers the wizard rather than the agent.
         run: >-
           uv run
           --with-requirements setup-requirements.txt
           --with pytest
           --with pytest-cov
+          --with fastapi
+          --with uvicorn
           pytest tests/unit
           --cov
           --cov-config=.coveragerc
@@ -153,8 +158,15 @@ jobs:
         run: uv sync --locked --group test
 
       - name: Run unit tests with coverage
+        # chronicle_setup comes in on the command line rather than as a path dep:
+        # tests/test_memory_setup_wizard.py loads backends/advanced/init.py, which
+        # imports it, but ../../extras is outside the Docker build context, so
+        # declaring it in pyproject.toml would break the image build. Same reasoning
+        # as the chronicle-wearable-sdk git source there.
         run: >-
-          uv run --group test pytest
+          uv run --group test
+          --with ../../extras/chronicle-setup
+          pytest
           --cov=advanced_omi_backend
           --cov-report=term-missing
           --cov-report=xml

backends/advanced/src/advanced_omi_backend/constants.py

@@ -4,6 +4,10 @@
 OMI_CHANNELS = 1
 OMI_SAMPLE_WIDTH = 2  # bytes (16‑bit)
 
+# A missing generated title must remain visibly machine-detectable. Never replace it
+# with transcript text or a plausible label such as "Recording"/"Conversation".
+TITLE_NOT_GENERATED = "[title not generated]"
+
 # Reserved diarization label for segments triaged as background/noise (TV, media,
 # ambient) rather than a real person. Used by the Data Audit speaker-triage flow:
 # applying it sets the segment's speaker to this label AND reclassifies it to a

backends/advanced/src/advanced_omi_backend/services/forced_alignment.py

@@ -109,3 +109,30 @@ async def align_audio_words(
         f"({data.get('aligned_segments')}/{data.get('total_segments')} segments aligned)"
     )
     return words
+
+
+def estimate_words_from_segment_timing(segments: List[Dict]) -> List[Dict]:
+    """Create monotonic word clocks when the neural aligner cannot align text.
+
+    Segment timestamps remain authoritative; this only distributes their words
+    uniformly inside each segment so Pyannote speaker turns can receive the existing
+    transcript text.  It is deliberately a last resort after forced alignment.
+    """
+    words: List[Dict] = []
+    for segment in segments:
+        start = float(segment.get("start", 0.0))
+        end = float(segment.get("end", start))
+        tokens = str(segment.get("text", "")).split()
+        if not tokens or end <= start:
+            continue
+        step = (end - start) / len(tokens)
+        for index, token in enumerate(tokens):
+            words.append(
+                {
+                    "word": token,
+                    "start": start + index * step,
+                    "end": start + (index + 1) * step,
+                    "confidence": 0.0,
+                }
+            )
+    return words

backends/advanced/src/advanced_omi_backend/services/memory/vault_verify.py

@@ -32,6 +32,11 @@
 from .vault_scaffold import VaultPathError, safe_vault_relative_path
 
 _H2_RE = re.compile(r"^##\s+(.+?)\s*$", re.MULTILINE)
+_DAY_DIGEST_RANGE_RE = re.compile(r"^###\s+(\d{2}:\d{2}–\d{2}:\d{2})\s+·", re.MULTILINE)
+_DAY_NOTE_RANGE_RE = re.compile(
+    r"^-\s+(?:\*\*)?(\d{2}:\d{2}–\d{2}:\d{2})(?:\*\*)?\s+·",
+    re.MULTILINE,
+)
 
 # Long-lived structured notes carry a stable spine plus the aggregation embed that
 # auto-lists their conversations. A note missing either is malformed forever.
@@ -148,6 +153,47 @@ def illegal_path_reason(rel: str) -> str:
     return ""
 
 
+def root_note_role_reason(
+    root: Path, rel: str, before: str | None, content: str
+) -> str:
+    """Why a changed root Markdown note is not a valid category hub.
+
+    Content notes live one folder deep.  Root Markdown is reserved for the thin hub
+    notes that make category wikilinks resolve and embed their matching Obsidian Base.
+    Organic categories remain open-ended, but they must be created as the complete
+    template/base/hub bundle rather than by dropping an entity or topic at the root.
+    """
+
+    path = Path(rel)
+    if len(path.parts) != 1 or path.suffix != ".md":
+        return ""
+
+    category = path.stem
+    if before is not None:
+        return (
+            "root Markdown files are category hubs, not captured-content notes. Do not "
+            "edit the hub; put durable content in its category folder (for a topic, "
+            f"`Topics/{category}.md`)."
+        )
+
+    template = root / "Templates" / f"{category} Template.md"
+    base = root / "Templates" / "Bases" / f"{category}.base"
+    is_complete_hub = (
+        template.is_file()
+        and base.is_file()
+        and f"# {category}" in content
+        and f"![[{category}.base]]" in content
+    )
+    if is_complete_hub:
+        return ""
+    return (
+        "root Markdown files are reserved for category hubs created as a matching "
+        "template/base/hub bundle. If this is a topic, move it to "
+        f"`Topics/{category}.md`; if it is a new recurring kind of thing, create the "
+        "category first and file the note under `<Category>/<Title>.md`."
+    )
+
+
 def _markdown_files(root: Path) -> Dict[str, str]:
     """Every readable ``*.md`` in the vault, keyed by vault-relative POSIX path."""
 
@@ -163,6 +209,52 @@ def _markdown_files(root: Path) -> Dict[str, str]:
     return out
 
 
+def verify_day_episode_ranges(note_path: Path, day_digest: str) -> List[Finding]:
+    """Require the Daily episode index to mirror the active timeline exactly.
+
+    A day can be analysed again after it was already written. The write agent used to
+    interpret "add only what is missing" literally: it appended a newly discovered
+    episode but retained stale time ranges for every existing episode. The write then
+    looked healthy even though the vault no longer represented the active run.
+
+    The semantic wording stays agentic, but the episode index is a source-backed
+    contract: one ordered bullet per supplied episode, with the exact range selected by
+    segmentation. Raw transcripts remain outside the vault.
+    """
+
+    expected = _DAY_DIGEST_RANGE_RE.findall(day_digest or "")
+    try:
+        note = note_path.read_text(encoding="utf-8")
+    except OSError:
+        note = ""
+
+    episodes_heading = re.search(r"^##\s+Episodes\s*$", note, re.MULTILINE)
+    if episodes_heading is None:
+        section = ""
+    else:
+        section_start = episodes_heading.end()
+        next_heading = _H2_RE.search(note, section_start)
+        section_end = next_heading.start() if next_heading else len(note)
+        section = note[section_start:section_end]
+    actual = _DAY_NOTE_RANGE_RE.findall(section)
+
+    if actual == expected:
+        return []
+
+    rel = "/".join(note_path.parts[-2:])
+    return [
+        Finding(
+            rel,
+            "episode_ranges",
+            "replace only the `## Episodes` section with exactly one chronological "
+            "bullet per supplied day episode, using each source range verbatim and "
+            "removing every stale or duplicate bullet. "
+            f"Expected {len(expected)} range(s): {', '.join(expected) or '(none)'}. "
+            f"Found {len(actual)}: {', '.join(actual) or '(none)' }.",
+        )
+    ]
+
+
 def verify_vault_changes(
     root: Path,
     before: Mapping[str, str],
@@ -232,6 +324,10 @@ def verify_vault_changes(
         if reason:
             findings.append(Finding(rel, "illegal_path", reason))
 
+        reason = root_note_role_reason(root, rel, was, content)
+        if reason:
+            findings.append(Finding(rel, "root_note_role", reason))
+
         reason = non_person_note_reason(rel)
         if reason:
             findings.append(Finding(rel, "not_a_person", reason))

backends/advanced/src/advanced_omi_backend/services/sse_publisher.py

@@ -14,14 +14,18 @@
 import time
 
 import redis
+import redis.asyncio as aioredis
 
-from advanced_omi_backend.redis_factory import create_sync_redis
+from advanced_omi_backend.redis_factory import create_async_redis, create_sync_redis
 
 logger = logging.getLogger(__name__)
 
 # Lazy-initialized sync Redis client (for RQ workers)
 _sync_redis: redis.Redis | None = None
 
+# Lazy-initialized async client, for callers already on an event loop.
+_async_redis: aioredis.Redis | None = None
+
 
 def _get_sync_redis() -> redis.Redis:
     """Get or create the sync Redis client."""
@@ -31,6 +35,18 @@ def _get_sync_redis() -> redis.Redis:
     return _sync_redis
 
 
+def _get_async_redis() -> aioredis.Redis:
+    """Get or create the async Redis client. Must be called from the loop."""
+    global _async_redis
+    if _async_redis is None:
+        _async_redis = create_async_redis(decode_responses=True)
+    return _async_redis
+
+
+def _message(event_type: str, data: dict) -> str:
+    return json.dumps({"event": event_type, "data": data, "timestamp": time.time()})
+
+
 def publish_sse_event(user_id: str, event_type: str, data: dict) -> None:
     """
     Publish an SSE event to the user's channel (sync, for RQ workers).
@@ -42,19 +58,28 @@ def publish_sse_event(user_id: str, event_type: str, data: dict) -> None:
     """
     try:
         r = _get_sync_redis()
-        message = json.dumps(
-            {
-                "event": event_type,
-                "data": data,
-                "timestamp": time.time(),
-            }
-        )
-        r.publish(f"sse:{user_id}", message)
+        r.publish(f"sse:{user_id}", _message(event_type, data))
     except Exception:
         # SSE publishing is best-effort — never fail the calling job
         logger.debug("Failed to publish SSE event %s", event_type, exc_info=True)
 
 
+async def publish_sse_event_async(user_id: str, event_type: str, data: dict) -> None:
+    """Publish an SSE event from a caller already on the event loop.
+
+    The sync client above must not be used there. Its first command on a dropped
+    connection reconnects inline — a blocking ``getaddrinfo`` and ``connect`` on
+    the loop thread, measured here at 3-5s per reconnect, during which nothing
+    else in the process runs.
+    """
+    try:
+        r = _get_async_redis()
+        await r.publish(f"sse:{user_id}", _message(event_type, data))
+    except Exception:
+        # SSE publishing is best-effort — never fail the calling handler
+        logger.debug("Failed to publish SSE event %s", event_type, exc_info=True)
+
+
 # Throttle state for publish_sse_event_throttled
 _last_publish: dict[str, float] = {}