refactor(daemon): reconcile the clocks into one daemon-owned scheduler (V2-15) (#2199)

## Why this matters

Four independent clocks used to run GAIA's periodic work — the UI
backend's `Scheduler`, the `gaia schedule` CLI, and the email sidecar's
two in-process clocks (`BriefingScheduler` #1918, `EmailJobScheduler`
#1919) — and **each died with the process that owned it**. A briefing or
scheduled send stopped firing the moment the UI/CLI/sidecar closed, so
"always-on" was false; worse, V2-6's idle-sidecar reaper had to stay
disabled (#2142) because reaping a sidecar silently killed its clock.

This lands the spine of the fix: **one daemon-owned clock**
(`gaia.daemon.scheduler`) that adopts the in-sidecar jobs **exactly
once** through a migration ledger. A re-run of the migration is a no-op,
an unschedulable job raises loudly, and a dropped job is detected loudly
(`assert_no_dropped`) instead of silently vanishing — the exact
regression the epic flagged as most likely here. Under daemon
supervision (`GAIA_DAEMON_SUPERVISED`, injected at spawn) the email
sidecar gates its own embedded clocks off; standalone / bare-integrator
/ `CustodyProvider` runs never see the var and keep them live, so
scheduling never silently stops for anyone the daemon isn't driving.

**Draft, and why:** the daemon clock + reconciliation + email gating are
complete and tested here. Rewiring `routers/schedules.py`,
`routers/goals.py`, and the `gaia schedule` CLI into thin daemon clients
— and landing schedule/task state in *host custody* (AC #2) — rides on
**#2153** (custody store), which is being built in parallel. The clock's
SQLite store is deliberately custody-agnostic so it swaps onto #2153's
host-custody store without touching this logic. Expect a rebase once
#2153 lands; the idle-reaper flip (AC #6) is gated behind that same
follow-up so gating lands before reaping, never after.

## Test plan

- [x] `python -m pytest tests/unit/test_daemon_scheduler.py
tests/unit/test_daemon_scheduler_migration.py` — clock fires
exactly-once, recurring re-arms, missing executor / executor exception
fail loudly (not dropped), atomic claim blocks double-fire; reconcile is
idempotent, dropped-job + unschedulable-job raise loudly, no double-run
when old+new paths coexist.
- [x] `python -m pytest
hub/agents/python/email/tests/test_daemon_migration.py` — **exactly-once
migration of the in-sidecar jobs** (the flagged regression): real
`schedule_store` one-shots + the daily briefing migrate once, a re-run
adopts nothing, a fired job is never re-migrated; supervision gate
detected only on the exact env value.
- [x] `python -m pytest
hub/agents/python/email/tests/test_email_schedule.py` — existing email
scheduler suite stays green (standalone parity, AC #4): 16 passed.
- [x] `python util/lint.py` — black + isort + flake8(F) clean on all
changed files (pre-existing unrelated failures noted below).

Pre-existing, unrelated to this change (identical on `main`):
`tests/unit/test_agent_sidecar_manager.py` and one
`test_daemon_agents_routes.py` case fail on Windows (`os.killpg` /
group-kill are Unix-only); `tests/unit/test_schedule_daemon.py` errors
on a missing `tomli_w` dep in this env.

Part of #2014
Closes #2156

Co-authored-by: Ovtcharov <kovtchar@amd.com>

Author: kovtcharov-amd

hub/agents/python/email/CHANGELOG.md

@@ -0,0 +1,34 @@
+# Changelog — `gaia-agent-email`
+
+All notable changes to the GAIA Email Triage agent package are recorded here.
+Format loosely follows [Keep a Changelog](https://keepachangelog.com/); the REST
+contract version is tracked separately as
+`gaia_agent_email.contract.SCHEMA_VERSION` (see `CONTRACT.md`).
+
+## [Unreleased]
+
+### Changed
+
+- **Daemon-supervised scheduling (V2-15, #2156).** When the GAIA daemon spawns
+  the sidecar it sets `GAIA_DAEMON_SUPERVISED=1`; in that mode the sidecar's two
+  embedded clocks — the daily `BriefingScheduler` (#1918) and the one-shot
+  `EmailJobScheduler` polling thread (#1919) — no longer start. The daemon owns
+  a single reconciled clock and drives those jobs itself, so a scheduled brief
+  or send now fires even with the web UI and CLI closed, and can no longer be
+  silently killed when an idle sidecar is reaped.
+
+  This is **additive and gated by supervision context, not a deletion**: a
+  standalone `gaia-agent-email serve`, a bare integrator, or a
+  `CustodyProvider` deployment never sees the env var and keeps both embedded
+  clocks live exactly as before. The frozen `/v1/email/*` REST contract and
+  `SCHEMA_VERSION` are unchanged.
+
+### Added
+
+- `gaia_agent_email.supervision.is_daemon_supervised()` — detects the daemon
+  supervision handshake (the env-var name is owned by core in
+  `gaia.daemon.constants`, so daemon and sidecar can never drift).
+- `gaia_agent_email.daemon_migration` — adapter that lifts the embedded clocks'
+  jobs (pending `schedule_store` one-shots + the enabled daily briefing) into
+  the daemon clock **exactly once** via the core reconciler's migration ledger,
+  and asserts no job is silently dropped in the process.

hub/agents/python/email/gaia_agent_email/agent.py

@@ -52,6 +52,7 @@ class never passes ``use_claude=True`` / ``use_chatgpt=True`` to
     AGENT_NAMESPACED_ID,
     ALL_SCOPES,
 )
+from gaia_agent_email.supervision import is_daemon_supervised
 from gaia_agent_email.tools.calendar_tools import CalendarToolsMixin
 from gaia_agent_email.tools.delete_tools import DeleteToolsMixin
 from gaia_agent_email.tools.followup_tools import FollowupToolsMixin
@@ -504,8 +505,18 @@ def __init__(self, config: Optional[EmailAgentConfig] = None):
             },
             poll_seconds=config.scheduler_poll_seconds,
         )
-        if config.start_scheduler:
+        # V2-15 (#2156): under daemon supervision the daemon drives one-shot
+        # jobs from its single reconciled clock, so the embedded polling thread
+        # stays off — two drivers over one store risks a double-fire. Standalone
+        # / bare integrator runs (no supervision env) keep the thread live.
+        if config.start_scheduler and not is_daemon_supervised():
             self._scheduler.start()
+        elif config.start_scheduler:
+            logger.info(
+                "Email agent under daemon supervision: embedded "
+                "EmailJobScheduler polling thread gated off (the daemon drives "
+                "scheduled send / snooze from its reconciled clock)."
+            )
 
     # -- Agent contract -----------------------------------------------------
 

hub/agents/python/email/gaia_agent_email/briefing.py

@@ -291,6 +291,13 @@ class BriefingScheduler:
     A failed run (mailbox disconnected, provider outage) is logged with its
     actionable message and the schedule continues; it is never retried
     silently or downgraded to a partial briefing.
+
+    Daemon supervision (V2-15, #2156): when the GAIA daemon spawns this sidecar
+    it drives the daily brief from its single reconciled clock, so
+    ``server.py`` does NOT start this in-process timer (see
+    :func:`gaia_agent_email.supervision.is_daemon_supervised`). Standalone /
+    bare-integrator runs are unaffected and keep this timer live — the seam the
+    #1371 dispatcher note below anticipated is now realized by the daemon.
     """
 
     def __init__(

hub/agents/python/email/gaia_agent_email/daemon_migration.py

@@ -0,0 +1,155 @@
+# Copyright(C) 2025-2026 Advanced Micro Devices, Inc. All rights reserved.
+# SPDX-License-Identifier: MIT
+"""Adapter: lift the email sidecar's in-process clock jobs into the daemon
+clock (V2-15, #2156).
+
+This is the email-specific half of the reconciliation. It reads the two
+embedded clocks' durable state and maps it to the daemon's hub-safe
+:class:`~gaia.daemon.scheduler.models.MigratableJob` vocabulary, then hands the
+batch to the core reconciler. Living in the hub package keeps the dependency
+direction legal: hub -> core is fine, and core never learns anything
+email-specific.
+
+Two sources fold in here:
+
+- ``EmailJobScheduler`` / ``schedule_store`` (#1919) — persistent one-shot jobs
+  (scheduled send, snooze) already keyed by a stable ``job_id``. Each pending
+  job becomes a one-shot :class:`MigratableJob`.
+- ``BriefingScheduler`` (#1918) — a recurring daily brief configured from env,
+  with no per-job row. When enabled it contributes a single recurring
+  :class:`MigratableJob` with a synthetic-but-stable ``source_job_id`` so a
+  re-run migrates it exactly once.
+
+The reconcile is idempotent: run it every time the daemon (re)adopts the email
+sidecar; the ledger makes the second and later passes no-ops.
+"""
+
+from __future__ import annotations
+
+from datetime import datetime
+from typing import List, Optional
+
+from gaia_agent_email import schedule_store
+from gaia_agent_email.briefing import (
+    BriefingScheduleConfig,
+    seconds_until_next_run,
+)
+
+from gaia.daemon.scheduler import (
+    KIND_ONE_SHOT,
+    KIND_RECURRING,
+    MigratableJob,
+    MigrationResult,
+    assert_no_dropped,
+    reconcile_jobs,
+)
+
+# Source names recorded in the daemon migration ledger. Stable strings — they
+# key the exactly-once guard, so they must never change once shipped.
+SOURCE_ONE_SHOT = "email:schedule_store"
+SOURCE_BRIEFING = "email:briefing"
+
+# The briefing has no per-job row; this fixed id gives it a stable ledger key so
+# re-adoption is idempotent. One daily brief per sidecar identity.
+BRIEFING_JOB_ID = "daily_inbox_briefing"
+
+
+def collect_one_shot_jobs(db) -> List[MigratableJob]:
+    """Every still-pending one-shot email job as a :class:`MigratableJob`."""
+    schedule_store.init_schema(db)
+    jobs: List[MigratableJob] = []
+    for row in schedule_store.list_jobs(db, status=schedule_store.STATUS_PENDING):
+        jobs.append(
+            MigratableJob(
+                source=SOURCE_ONE_SHOT,
+                source_job_id=row["job_id"],
+                kind=KIND_ONE_SHOT,
+                fire_at=row["due_at"],
+                payload={
+                    "kind": row["kind"],
+                    "mailbox": row["mailbox"],
+                    "payload": row["payload"],
+                },
+            )
+        )
+    return jobs
+
+
+def collect_briefing_job(
+    config: BriefingScheduleConfig, *, now: Optional[datetime] = None
+) -> List[MigratableJob]:
+    """The daily briefing as a recurring :class:`MigratableJob`, or [] when off.
+
+    A disabled briefing contributes nothing — matching the embedded scheduler,
+    which creates no task when disabled. The first ``fire_at`` is the next local
+    occurrence of the configured time; the interval is a fixed 24h.
+    """
+    config.validate()
+    if not config.enabled:
+        return []
+    reference = now or datetime.now()
+    delay = seconds_until_next_run(config.time_of_day, reference)
+    return [
+        MigratableJob(
+            source=SOURCE_BRIEFING,
+            source_job_id=BRIEFING_JOB_ID,
+            kind=KIND_RECURRING,
+            interval_seconds=86400,
+            fire_at=reference.timestamp() + delay,
+            payload={
+                "time_of_day": config.time_of_day,
+                "max_messages": config.max_messages,
+            },
+        )
+    ]
+
+
+def migrate_email_clocks(
+    db,
+    *,
+    briefing_config: Optional[BriefingScheduleConfig] = None,
+    now: Optional[datetime] = None,
+    verify_no_dropped: bool = True,
+) -> MigrationResult:
+    """Adopt both embedded email clocks into the daemon clock, exactly once.
+
+    ``db`` is a ``DatabaseMixin`` handle open on the daemon clock's store (the
+    same file the daemon drives). Pass ``briefing_config`` to fold the daily
+    brief in; omit it to migrate only the one-shot jobs.
+
+    When ``verify_no_dropped`` is set (the default) the one-shot batch is
+    checked with :func:`assert_no_dropped` after the pass — every pending email
+    job must have reached the ledger, or a :class:`DroppedJobError` is raised so
+    the migration fails loudly instead of silently losing a scheduled send.
+    """
+    one_shot = collect_one_shot_jobs(db)
+    briefing = (
+        collect_briefing_job(briefing_config, now=now)
+        if briefing_config is not None
+        else []
+    )
+    result = reconcile_jobs(db, one_shot + briefing)
+
+    if verify_no_dropped:
+        assert_no_dropped(
+            db,
+            source=SOURCE_ONE_SHOT,
+            source_job_ids=[j.source_job_id for j in one_shot],
+        )
+        if briefing:
+            assert_no_dropped(
+                db,
+                source=SOURCE_BRIEFING,
+                source_job_ids=[BRIEFING_JOB_ID],
+            )
+    return result
+
+
+__all__: List[str] = [
+    "BRIEFING_JOB_ID",
+    "SOURCE_BRIEFING",
+    "SOURCE_ONE_SHOT",
+    "collect_briefing_job",
+    "collect_one_shot_jobs",
+    "migrate_email_clocks",
+]

hub/agents/python/email/gaia_agent_email/scheduler.py

@@ -24,6 +24,13 @@
 ``start()``-ing the thread. The store and executors don't change — only the
 driver does. Kept email-scoped and minimal on purpose.
 
+Daemon supervision (V2-15, #2156): under the GAIA daemon this thread is NOT
+started (see :func:`gaia_agent_email.supervision.is_daemon_supervised`); the
+daemon drives these jobs from its single reconciled clock after adopting them
+via :func:`gaia_agent_email.daemon_migration.migrate_email_clocks`. Standalone
+runs keep the thread — the seam above, realized. The atomic ``claim_job`` guard
+means even if both drivers briefly poll the same store, each job fires once.
+
 Fail-loudly contract: an executor failure marks the job ``failed`` with the
 error message persisted on the row and logs at ERROR — a firing send must
 never silently swallow a send failure. A job whose kind has no registered

hub/agents/python/email/gaia_agent_email/server.py

@@ -81,6 +81,7 @@ def build_app():
     from gaia_agent_email.briefing import BriefingScheduleConfig, BriefingScheduler
     from gaia_agent_email.connector_routes import router as connector_router
     from gaia_agent_email.contract import SCHEMA_VERSION
+    from gaia_agent_email.supervision import is_daemon_supervised
 
     # Daily inbox briefing (#1608) — env config is read at build time so an
     # invalid value aborts startup loudly, not at the first scheduled fire.
@@ -89,6 +90,18 @@ def build_app():
 
     @asynccontextmanager
     async def lifespan(_app):
+        # V2-15 (#2156): under daemon supervision the daemon drives the brief
+        # from its single reconciled clock, so the embedded clock stays dark —
+        # running both over one store is a double-run. Standalone / bare
+        # integrator runs (no supervision env) keep the embedded clock live.
+        if is_daemon_supervised():
+            log.info(
+                "Email sidecar under daemon supervision: embedded "
+                "BriefingScheduler gated off (the daemon drives the daily "
+                "brief from its reconciled clock)."
+            )
+            yield
+            return
         scheduler = BriefingScheduler(briefing_config)
         scheduler.start()
         try:

hub/agents/python/email/gaia_agent_email/supervision.py

@@ -0,0 +1,45 @@
+# Copyright(C) 2025-2026 Advanced Micro Devices, Inc. All rights reserved.
+# SPDX-License-Identifier: MIT
+"""Daemon-supervision detection for the email sidecar (V2-15, #2156).
+
+When the GAIA daemon spawns this sidecar it sets
+``GAIA_DAEMON_SUPERVISED=1`` in the environment. In that mode the daemon owns
+the clock: it drives the briefing and one-shot jobs from its single reconciled
+scheduler, so the sidecar's OWN embedded schedulers (``BriefingScheduler`` #1918,
+``EmailJobScheduler`` #1919) must NOT also run — two clocks over one store is the
+double-run this reconciliation exists to prevent.
+
+A sidecar started any other way — a bare integrator, a standalone
+``gaia-agent-email serve``, an embedded ``CustodyProvider`` deployment — never
+sees the var and keeps its embedded clocks live. The check is a supervision
+*context* test, deliberately NOT a deletion, so standalone scheduling behavior
+(and its test suite) is untouched.
+
+The env-var NAME is owned by core (``gaia.daemon.constants``) so the daemon that
+sets it and the sidecar that reads it can never drift. Importing a core constant
+from a hub package is allowed; core never imports a hub wheel.
+"""
+
+from __future__ import annotations
+
+import os
+from typing import Mapping, Optional
+
+from gaia.daemon.constants import (
+    DAEMON_SUPERVISION_ENABLED_VALUE,
+    DAEMON_SUPERVISION_ENV_VAR,
+)
+
+
+def is_daemon_supervised(environ: Optional[Mapping[str, str]] = None) -> bool:
+    """True when the daemon is driving this sidecar's clock.
+
+    Only the exact enabled value counts — any other value (including a stray
+    empty string) means "not supervised", so a misconfigured env never
+    silently disables the embedded clocks a standalone run depends on.
+    """
+    env = os.environ if environ is None else environ
+    return env.get(DAEMON_SUPERVISION_ENV_VAR) == DAEMON_SUPERVISION_ENABLED_VALUE
+
+
+__all__ = ["is_daemon_supervised"]