schedule_store.py: acquire_due_jobs has no worker mutex lock → concurrent scheduler tick double-fires (CWE-362)

Component: primitives/schedule_store.py — acquire() (station v0.97).

The defect.
The Schedule Store guarantees that a live lock prevents concurrent executions of the same schedule (primitives/schedule_store.py:267-269 "A LIVE lock means another run of THIS schedule is in progress. NEITHER 'skip' NOR 'queue' may run concurrently — the lock exists precisely to prevent that").

However, acquire() at lines 255–290 loads the schedule record, inspects lock, and writes the newly acquired lock across non-atomic phases without mutex serialization. _SCHEDULE_LOCK was added to other state stores but omitted here.

When two concurrent scheduler ticks occur (e.g. overlapping timer intervals or simultaneous HTTP /api/schedules/tick invocations under ThreadingMixIn), both threads observe lock = None simultaneously, write their respective locks, and both return ok = True, causing unattended scheduled workflows to execute twice.

Reproduction (standalone harness).

import os, shutil, tempfile, json, time, threading

try:
    from primitives import schedule_store as ss
except ImportError:  # Workbench layout
    from workbench.primitives import schedule_store as ss

ws = tempfile.mkdtemp(prefix="rc_sched_race_")

sched_id = "sch_sync_orders_01"
rec = {
    "id": sched_id,
    "title": "Sync Orders to ERP",
    "interval_minutes": 15,
    "concurrency": "skip",
    "workflow_id": "wf_orders_sync",
    "enabled": True,
    "last_run_at": "2026-08-15T22:00:00Z"
}
ss._save(ws, rec)

# Note: Thread-isolated tmp path isolates filesystem writes in test harness without altering acquire() logic
orig_save = ss._save
def safe_save(ws, r):
    p = ss.path_for(ws, r["id"])
    tmp = f"{p}.{threading.get_ident()}.tmp"
    with open(tmp, "w") as fh:
        json.dump(r, fh, indent=2, sort_keys=True)
    os.replace(tmp, p)
    return r
ss._save = safe_save

barrier = threading.Barrier(2)
acquire_results = []

def worker(tid):
    barrier.wait()
    ok, updated_rec, reason = ss.acquire(ws, sched_id, run_id=f"run_worker_{tid}")
    acquire_results.append({
        "worker": tid,
        "ok": ok,
        "run_id": updated_rec.get("lock", {}).get("run_id") if updated_rec else None
    })

t1 = threading.Thread(target=worker, args=(1,))
t2 = threading.Thread(target=worker, args=(2,))
t1.start(); t2.start(); t1.join(); t2.join()

print("Both Acquired OK (Double-Fire):", [r["ok"] for r in acquire_results])
print("Distinct Run IDs Minted:", len({r["run_id"] for r in acquire_results if r["run_id"]}) == 2)

shutil.rmtree(ws)

Expected raw output on vulnerable code:

Both Acquired OK (Double-Fire): [True, True]
Distinct Run IDs Minted: True

Scope (stated honestly).

• Affects unattended scheduled workflows (disposition="auto") triggered during overlapping or concurrent tick intervals.
• This is a state-machine serialization defect in lock acquisition.

Fix.
Wrap acquire() and release() operations in a module-level _SCHEDULE_LOCK mutex:

--- a/primitives/schedule_store.py
+++ b/primitives/schedule_store.py
@@ -250,6 +250,9 @@
 # ── locking ──────────────────────────────────────────────────────────────────
 
+_SCHEDULE_LOCK = threading.Lock()
+
 def acquire(ws, schedule_id, *, now=None, run_id=None):
     """(ok, rec, reason). reason is SKIPPED_OVERLAP when a live run holds it,
     STALE_LOCK_RECLAIMED when a dead run's lock was reclaimed."""
+    with _SCHEDULE_LOCK:
+        now = now or _now()
+        rec = load(ws, schedule_id)
+        if not rec:
+            return False, None, "unknown schedule"
+        # ... critical section: check lock, set lock, persist ...
+        rec["last_tick_at"] = _iso(now)
+        return True, _save(ws, rec), reason

Classification: CWE-362