Skip to content

symfonic.diagnostics.checks.check_consolidate_schedule

check_consolidate_schedule

Consolidate-beat-schedule audit (v7.16, item 5).

Detects the hardcoded "default" tenant trap in the adopter's Celery beat schedule. Scaffolded projects ship with::

"consolidate-memory-default": {
    "task": "<pkg>.tasks.consolidate_memory_task",
    "schedule": crontab(hour=3, minute=0),
    "args": ("__all__",),
}

A common edit -- copy-pasted from older guides -- changes "__all__" to "default", which silently restricts nightly consolidation to a single tenant. Multi-tenant adopters then watch their consolidation metrics flatline.

Detection strategy: runtime introspection of any registered Celery app.

  1. Try the conventional app.worker import path.
  2. Fall back to scanning sys.modules for any module exposing a :class:celery.Celery instance.
  3. If no Celery is in play, skip silently -- not every adopter uses Celery, and we don't want to flag a check that doesn't apply.

For each beat entry, inspect the args tuple; if any element is the literal string "default", emit a WARN. We deliberately avoid clever schema inference -- adopters with non-default args shapes are responsible for their own audit.

check_consolidate_schedule async 

check_consolidate_schedule(
    agent: SymfonicAgent,
) -> list[CheckResult]

Audit the Celery beat schedule for the hardcoded-tenant trap.

Source code in src/symfonic/diagnostics/checks/check_consolidate_schedule.py 
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
async def check_consolidate_schedule(
    agent: SymfonicAgent,  # noqa: ARG001 -- uniform signature
) -> list[CheckResult]:
    """Audit the Celery beat schedule for the hardcoded-tenant trap."""
    celery_app = _discover_celery_app()
    if celery_app is None:
        # No Celery in this process -- silent skip per architect verdict.
        return []

    schedule = _extract_beat_schedule(celery_app)
    if not schedule:
        return [
            CheckResult(
                name="consolidate_schedule.no_beat",
                severity=Severity.INFO,
                message=(
                    "Celery app discovered but no beat_schedule registered "
                    "-- nightly consolidation will not run automatically."
                ),
                fix_hint=(
                    "Register a beat schedule entry for "
                    "consolidate_memory_task with args=('__all__',) to "
                    "consolidate every tenant."
                ),
            ),
        ]

    results: list[CheckResult] = []
    for entry_name, entry in schedule.items():
        args = _extract_args(entry)
        if any(_is_default_tenant(a) for a in args):
            results.append(
                CheckResult(
                    name="consolidate_schedule.hardcoded_default",
                    severity=Severity.WARN,
                    message=(
                        f"Beat entry {entry_name!r} schedules a task with "
                        "args containing the literal 'default' -- nightly "
                        "consolidation will run for that single tenant only."
                    ),
                    fix_hint=(
                        "Change args=('default',) to args=('__all__',) to "
                        "consolidate every tenant."
                    ),
                ),
            )
    return results