Fix detached DataSource and expired sensors when enqueuing forecast jobs

[Copilot]

get_or_create_source() only flushes a new DataSource — it never commits. When the API request context tears down without an explicit commit, the data source row is rolled back. The RQ worker then fails to find it by ID. A secondary issue: db.session.commit() expires all session objects, causing sensor references pickled into RQ jobs (future_regressors, past_regressors, sensor, sensor_to_save) to raise DetachedInstanceError on the worker.

Fixes

• train_predict.py

  • db.session.merge(self.data_source) + db.session.commit() before enqueuing jobs — ensures the data source row exists in the DB when the worker runs
  • db.session.refresh(sensor) calls after commit to re-hydrate expired sensor objects before RQ serialises the job
  • New _reattach_if_needed() static helper consolidates the repeated merge-if-detached-or-expired pattern used across sensor, sensor_to_save, future_regressors, and past_regressors in run_cycle()
  • Fixed run_wrap_up(): Job.fetch() requires an explicit connection argument (was raising TypeError)
  • Return value of run() changed from a bare job ID string to {"job_id": ..., "n_jobs": ...}

• api/v3_0/sensors.py — consume pipeline_returns["job_id"]

• cli/data_add.py — use local as_job variable (was reading parameters.get("as_job")); consume pipeline_returns["n_jobs"]

Tests

• Regression test test_data_source_committed_before_forecast_jobs: demonstrates the flush-without-commit bug (source disappears after rollback) and verifies the merge+commit fix makes it findable in a fresh query
• Added multi-cycle parametrize case (retrain-frequency: PT12H, 2 cycles, as_job=True) — previously untested code path for wrap-up jobs
• Fixed forecast-count assertion to multiply by n_cycles; switched search_beliefs to most_recent_beliefs_only=False to count all saved beliefs across cycles

Original prompt

In #2011 we were facing a detached data source (representing the forecaster) that was not being committed by the server when the forecast trigger API was valued. Subsequently, the worker could not find the data source in the db and failed the forecasting job. A db session.merge(self.data_source) fixed that, but does not seem to address the root cause for the data source to become detached. Additionally, it lead to a test failing on a detached sensor (a future regressor), which is puzzling me.

You must:

• have the test specialist debug the issue using debug statements, not just inspecting code.
• have the test specialist add a test that reproduces the original data source issue if the db.session.merge(self.data_source) is not there. (You may need to locally revert the relevant commit to check.)
• verify all tests pass before stopping the session.

Custom agent used: review-lead
Orchestrates team of specialist agents for comprehensive code reviews and synthesizes their findings into unified recommendations

---

💬 Send tasks to Copilot coding agent from Slack and Teams to turn conversations into code. Copilot posts an update in your thread when it's finished.

[read-the-docs-community]

Documentation build overview

📚 flexmeasures | 🛠️ Build #31853186 | 📁 Comparing cbff3b4 against latest (f1707fc)

🔍 Preview build

Show files changed (3 files in total): 📝 3 modified | ➕ 0 added | ➖ 0 deleted

File	Status
genindex.html	📝 modified
_autosummary/flexmeasures.data.models.forecasting.pipelines.train_predict.html	📝 modified
api/v3_0.html	📝 modified