High — pipeline_name stores the synthetic job label, not the pipeline name.

Flagged independently by the code-review, silent-failure, comment, and test agents. task_args[6] on this path is the name built in SchedulerHelper.add_or_update_job as f"Pipeline job-{pipeline.id}" (helper.py:68 → helper.py:60). So this column is always persisted as "Pipeline job-<uuid>" — effectively a prefixed duplicate of pipeline_id, never the user-facing name the field name implies. (The other producer, piepline_api_execution_views.py, does pass the real pipeline.pipeline_name at index 6, but it dispatches via celery_app.send_task and never mirrors — so the mirror only ever sees the synthetic label.)

The broad except can't catch this (it's a valid string write) and the unit test feeds _NAME="Nightly ETL", hiding it.

Fix: source the real name from the Pipeline object and pass it explicitly, OR drop pipeline_name from this inert slice rather than persisting a misleading value. At minimum, correct the layout comment and the test fixture so the false contract isn't pinned.

Fixed in b856507 — the mirror upsert moved to SchedulerHelper._schedule_task_job, which holds the Pipeline object and now passes pipeline_name=pipeline.pipeline_name (the real user-facing name). create_or_update_periodic_task no longer mirrors / parses args. Added a helper-wiring test asserting the real name flows (not the synthetic Pipeline job-<id> label).

Medium — silent no-op desync; asymmetric with the upsert path.

filter(pipeline_id=...).update(...) (and the .delete() sibling at line 66) silently match zero rows if the mirror row doesn't exist — e.g. a pipeline scheduled before this code shipped, or one whose update_or_create upsert previously failed and was swallowed. The Beat PeriodicTask stays correct, but the mirror diverges with no signal, and because enable/disable only updates-if-present it can never self-heal: a pipeline that is only ever paused/resumed is never backfilled. The future scheduler that reads enabled+next_run_at would then silently never fire (or fire a paused) pipeline.

This asymmetry (update_or_create on create vs filter().update() here) is the mechanism. Acceptable for the inert slice, but worth hardening before the table is read.

Fix: standardize the mutating paths on update_or_create semantics so every mirror mutation converges regardless of prior row existence, OR check the .update()/.delete() rowcount and logger.info when it's 0 so the gap is observable.

Addressed in b856507 — _mirror_periodic_schedule_set_enabled now checks the .update() rowcount and logger.infos when it matches 0 (a pre-existing/unmirrored schedule), so the gap is observable. Full self-heal/backfill of pre-existing schedules lands with the scheduler that reads the table (②b) — agreed it's not needed while the table is inert. +test asserting the 0-row log.

Medium — updated_at (auto_now=True) is not bumped by the enable/disable path.

Django's auto_now fires only on Model.save()/update_or_create, not on queryset .update(). _mirror_periodic_schedule_set_enabled (tasks.py:56) uses .filter().update(enabled=...), so pause/resume changes enabled without advancing updated_at. The upsert path bumps it correctly, making updated_at an unreliable "last changed" signal that silently misses enable/disable mutations.

Fix: add "updated_at": timezone.now() to the .update(...) call in _mirror_periodic_schedule_set_enabled, or add a one-line comment that updated_at intentionally tracks only definition upserts.

Fixed in b856507 — .update(enabled=..., updated_at=timezone.now()); queryset .update() doesn't fire auto_now, so this keeps updated_at accurate on pause/resume. Dev-tested: updated_at advances on disable.

Medium — positional task_args coupling silently degrades to placeholder data.

The task_args[N] if len(task_args) > N else <default> guards prevent an IndexError crash (good), but they convert a contract drift into silent placeholder data: if a future change reorders or shortens task_args (note the TODO: Remove unused args with a migration on this very contract), the mirror quietly stores organization_id="", workflow_id=None, etc., with no log. Worse, since this runs inside _mirror_periodic_schedule_upsert's broad except, even an off-by-one IndexError would be swallowed and silently drop the mirror row — and the short-list branches have zero test coverage.

Fix: create_or_update_periodic_task is called from _schedule_task_job, which already has workflow_id/organization_id/name as named locals (helper.py:41-44). Pass them as named params instead of re-parsing the serialized positional list. If positional parsing must stay, logger.warning when len(task_args) is shorter than expected.

Fixed in b856507 — dropped the positional task_args[N] parsing entirely. The upsert is now driven from _schedule_task_job with named locals (workflow_id, organization_id, pipeline.pipeline_name, pipeline.pk), so a future task_args reorder/shorten can't silently degrade the mirror to placeholders.

Medium — cron_string is unvalidated free text, against this file's own convention.

Every sibling model encodes its domain bounds as a DB constraint (PgQueueMessage.priority range, PgBatchDedup.batch_index >= 0, PgBarrierState.expires_at > created_at), yet cron_string is a bare TextField() with no validator and no length cap — despite the Beat path requiring exactly 5 whitespace-separated fields (cron_string.split() at tasks.py:85). A malformed value mirrors silently and only fails later in the future scheduler tick, far from the write site.

Fix: a cron CheckConstraint is impractical, so validate-at-write is the pragmatic fit: check the 5-field split at the helper boundary (where the cron is already in hand) before mirroring, and/or add a MaxLengthValidator/length cap. Don't over-engineer a regex into the schema.

The structural 5-field validation already exists upstream and runs before the mirror: create_or_update_periodic_task does minute, hour, dom, moy, dow = cron_string.split() (raises ValueError on a non-5-field value), and the mirror upsert now runs after that in the same helper call — so a malformed cron never reaches the table. Full cron semantics are validated by croniter in ②b (the reader/next-run computation), where a parse failure is actionable. Kept TextField per your 'don't over-engineer a regex into the schema' note. Added a comment to that effect.

Medium — test gaps (pr-test-analyzer).

The happy-path contract is well covered and follows the DB-free pg_queue/tests/test_producer.py convention, but several gaps could hide real bugs once the table goes live:

1. Failure-swallowing is tested for the upsert only. Add tests that make sched.objects.filter(...).update / .delete raise and assert disable_task/enable_task/delete_periodic_task don't propagate and still call PipelineProcessor.update_pipeline — this is the central "never break Beat" guarantee for 2 of 3 helpers, currently unverified (tasks.py:54-71).
2. Short task_args — every test uses the full 7-element list, so the else fallbacks (tasks.py:116-118) and the swallowed-IndexError path are never exercised. Call with []/a 2-element list and assert the defaults land.
3. enabled=False propagation through create_or_update_periodic_task (real callers pass enabled=pipeline.active).
4. Schema drift is invisible — the model is fully mocked, so a field-name mismatch between the upsert kwargs and the new model/migration passes green and fails only at runtime (then gets swallowed). Add one @pytest.mark.django_db round-trip, or a CI makemigrations --check gate.

Addressed in b856507: added (1) failure-swallow tests for disable/enable/delete asserting they don't propagate AND still call update_pipeline; (3) enabled=False upsert test; the 0-row-match log test. (2) short-task_args is now moot — the positional parsing was removed (named params from the helper). (4) schema-drift: makemigrations --check is clean and the real-DB lifecycle dev-test (CREATE→DISABLE→ENABLE→DELETE) exercises the actual model round-trip; a django_db unit round-trip needs the gated unit-backend rig group (postgres) which doesn't list scheduler/tests yet — noted as the same gap prior pg_queue PRs have.

Low — helper params typed Any hide a known contract.

The upstream caller already guarantees these are strings: str(pipeline.pk), str(workflow_id), str(name), and organization_id from UserContext.get_organization_identifier() (helper.py:48-61). Tightening to pipeline_id: str, organization_id: str, workflow_id: str | None, pipeline_name: str (keep cron_string: str, enabled: bool) costs nothing and documents that the mirror receives stringified UUIDs the UUIDField then coerces.

Fixed in b856507 — mirror_periodic_schedule_upsert params tightened to pipeline_id: str, organization_id: str, workflow_id: str | None, pipeline_name: str (+ cron_string: str, enabled: bool).

Low — fixture pins a contract production never produces.

_NAME = "Nightly ETL" at index 6 (asserted as pipeline_name) masks the High finding on tasks.py:118 — the real value flowing through SchedulerHelper is "Pipeline job-<uuid>". Use a representative fixture, e.g. _NAME = f"Pipeline job-{_PIPELINE_ID}", so the test reflects reality rather than enshrining a false expectation.

Resolved by the High fix (b856507): the mirror now stores the real pipeline.pipeline_name, so a representative name in the fixture is now the correct expectation, not a false one. The new helper-wiring test explicitly passes the synthetic Pipeline job-<id> label as the PeriodicTask name and asserts the mirror still records the real pipeline name — pinning the right contract.

Nit — index name drops the table prefix used by siblings.

Sibling indexes use the full table name as prefix (pg_queue_message_dequeue_idx, pg_barrier_expires_idx); this one abbreviates to pg_sched_due_idx. Cosmetic only — rename to pg_periodic_schedule_due_idx for consistency if you're already churning the migration, otherwise leave it (a rename costs a migration).

Fixed in b856507 — renamed to pg_periodic_schedule_due_idx (matches the sibling pg_queue_message_dequeue_idx / pg_barrier_expires_idx convention). Migration 0008 regenerated in place (not yet merged, so no extra migration).

str(workflow_id) converts a Python None into the string literal "None", which is truthy. The guard inside mirror_periodic_schedule_upsert — workflow_id or None — therefore never triggers for a None input, so Django's UUIDField receives "None", rejects it as an invalid UUID, and the whole upsert is silently swallowed. The function's own type annotation declares workflow_id: str | None, and PgPeriodicSchedule.workflow_id is null=True, so passing None is the intended and safe path.

This is a comment left during a code review.
Path: backend/scheduler/helper.py
Line: 69-72

Comment:
`str(workflow_id)` converts a Python `None` into the string literal `"None"`, which is truthy. The guard inside `mirror_periodic_schedule_upsert` — `workflow_id or None` — therefore never triggers for a `None` input, so Django's `UUIDField` receives `"None"`, rejects it as an invalid UUID, and the whole upsert is silently swallowed. The function's own type annotation declares `workflow_id: str | None`, and `PgPeriodicSchedule.workflow_id` is `null=True`, so passing `None` is the intended and safe path.

```suggestion
        mirror_periodic_schedule_upsert(
            pipeline_id=str(pipeline.pk),
            organization_id=organization_id or "",
            workflow_id=str(workflow_id) if workflow_id is not None else None,
```

How can I resolve this? If you propose a fix, please make it concise.