fix: converge Today session conflicts across devices (Closes #1030)

tests/test_today_session_handoff_ui.py

@@ -15,9 +15,13 @@ def test_dashboard_packages_a_mobile_today_session_handoff():
 
     assert 'id="today-session-handoff"' in html
     assert 'id="continue-today-session"' in html
+    assert 'id="today-session-sync-status"' in html
     assert 'aria-live="polite"' in html
     assert "createTodaySessionSync" in today_bundle
     assert "attachTodaySessionHandoff" in today_bundle
+    assert "Session syncing…" in today_bundle
+    assert "Session offline · will retry." in today_bundle
+    assert "Session continued on another device." in today_bundle
     assert "todaySessionSync = attachTodaySessionHandoff" in dashboard
     assert "todaySessionSync?.publish(snapshot)" in dashboard
     assert "min-height:44px" in css

tests/test_today_session_sync.py

@@ -119,3 +119,199 @@ const sync = createTodaySessionSync({
             "running": True,
         },
     }
+
+
+def test_publish_conflict_adopts_remote_owner_and_pauses_once():
+    result = run_node(
+        r"""
+const calls = [];
+const owned = {revision:2, device_id:'desktop-a', identity:'issue:r:42:', elapsed_ms:1000, running:true};
+const remote = {...owned, revision:3, device_id:'phone-b', elapsed_ms:2500};
+let pauses = 0;
+const transfers = [];
+const statuses = [];
+let reads = 0;
+const sync = createTodaySessionSync({
+  getDeviceId:()=> 'desktop-a',
+  fetchJson:async (url, options={}) => {
+    calls.push(options.method || 'GET');
+    if (options.method === 'PATCH') {
+      const error = new Error('session changed');
+      error.status = 409;
+      error.code = 'session_changed';
+      throw error;
+    }
+    reads += 1;
+    return reads === 1 ? owned : remote;
+  },
+  timer:{
+    snapshot:()=>({identity:owned.identity, elapsed_ms:1500, running:true}),
+    pause:()=>{pauses += 1;},
+  },
+  onTransferred:session=>transfers.push(session.device_id),
+  onStatus:status=>statuses.push(status),
+});
+(async()=>{
+  await sync.refresh();
+  await sync.publish();
+  await sync.refresh();
+  process.stdout.write(JSON.stringify({calls, pauses, transfers, statuses, session:sync.session()}));
+})().catch(error=>{console.error(error);process.exit(1);});
+"""
+    )
+
+    assert result["calls"] == ["GET", "PATCH", "GET", "GET"]
+    assert result["session"]["device_id"] == "phone-b"
+    assert result["pauses"] == 1
+    assert result["transfers"] == ["phone-b"]
+    assert "conflict" in result["statuses"]
+
+
+def test_rapid_publishes_are_single_flight_and_coalesce_latest_snapshot():
+    result = run_node(
+        r"""
+const owned = {revision:1, device_id:'desktop-a', identity:'issue:r:42:', elapsed_ms:0, running:true};
+const bodies = [];
+let active = 0;
+let peak = 0;
+let releaseFirst;
+let patchCount = 0;
+const firstPatch = new Promise(resolve=>{releaseFirst=resolve;});
+const sync = createTodaySessionSync({
+  getDeviceId:()=> 'desktop-a',
+  fetchJson:async (url, options={}) => {
+    if (!options.method) return owned;
+    active += 1;
+    peak = Math.max(peak, active);
+    patchCount += 1;
+    const body = JSON.parse(options.body);
+    bodies.push(body);
+    if (patchCount === 1) await firstPatch;
+    active -= 1;
+    return {...owned, revision:1 + patchCount, elapsed_ms:body.elapsed_ms, running:body.running};
+  },
+});
+(async()=>{
+  await sync.refresh();
+  const first = sync.publish({identity:owned.identity, elapsed_ms:1000, running:true});
+  await Promise.resolve();
+  const middle = sync.publish({identity:owned.identity, elapsed_ms:2000, running:false});
+  const latest = sync.publish({identity:owned.identity, elapsed_ms:3000, running:true});
+  releaseFirst();
+  await Promise.all([first, middle, latest]);
+  process.stdout.write(JSON.stringify({peak, bodies, session:sync.session()}));
+})().catch(error=>{console.error(error);process.exit(1);});
+"""
+    )
+
+    assert result["peak"] == 1
+    assert [body["elapsed_ms"] for body in result["bodies"]] == [1000, 3000]
+    assert result["session"]["elapsed_ms"] == 3000
+
+
+def test_remote_conflict_discards_queued_local_publishes_until_explicit_claim():
+    result = run_node(
+        r"""
+const owned = {revision:2, device_id:'desktop-a', identity:'issue:r:42:', elapsed_ms:1000, running:true};
+const remote = {...owned, revision:3, device_id:'phone-b', elapsed_ms:2500};
+const calls = [];
+let reads = 0;
+let releaseConflict;
+const conflictReady = new Promise(resolve=>{releaseConflict=resolve;});
+const sync = createTodaySessionSync({
+  getDeviceId:()=> 'desktop-a',
+  fetchJson:async (url, options={}) => {
+    calls.push(options.method || 'GET');
+    if (options.method === 'PATCH') {
+      await conflictReady;
+      const error = new Error('session changed');
+      error.status = 409;
+      throw error;
+    }
+    reads += 1;
+    return reads === 1 ? owned : remote;
+  },
+  timer:{pause:()=>{}},
+});
+(async()=>{
+  await sync.refresh();
+  const stale = sync.publish({identity:owned.identity, elapsed_ms:1500, running:true});
+  await Promise.resolve();
+  const queued = sync.publish({identity:owned.identity, elapsed_ms:2000, running:true});
+  releaseConflict();
+  await Promise.all([stale, queued]);
+  process.stdout.write(JSON.stringify({calls, session:sync.session()}));
+})().catch(error=>{console.error(error);process.exit(1);});
+"""
+    )
+
+    assert result["calls"] == ["GET", "PATCH", "GET"]
+    assert result["session"]["device_id"] == "phone-b"
+
+
+def test_transient_failure_retains_latest_intent_for_retry():
+    result = run_node(
+        r"""
+const owned = {revision:4, device_id:'desktop-a', identity:'issue:r:42:', elapsed_ms:1000, running:true};
+const bodies = [];
+const statuses = [];
+let attempts = 0;
+const sync = createTodaySessionSync({
+  getDeviceId:()=> 'desktop-a',
+  fetchJson:async (url, options={}) => {
+    if (!options.method) return owned;
+    attempts += 1;
+    bodies.push(JSON.parse(options.body));
+    if (attempts === 1) {
+      const error = new Error('temporarily unavailable');
+      error.status = 503;
+      throw error;
+    }
+    return {...owned, revision:5, elapsed_ms:bodies.at(-1).elapsed_ms};
+  },
+  onStatus:status=>statuses.push(status),
+});
+(async()=>{
+  await sync.refresh();
+  const first = await sync.publish({identity:owned.identity, elapsed_ms:4500, running:true});
+  const retried = await sync.publish();
+  process.stdout.write(JSON.stringify({first, retried, bodies, statuses}));
+})().catch(error=>{console.error(error);process.exit(1);});
+"""
+    )
+
+    assert result["first"] is None
+    assert [body["elapsed_ms"] for body in result["bodies"]] == [4500, 4500]
+    assert result["retried"]["revision"] == 5
+    assert "offline" in result["statuses"]
+
+
+def test_claim_transport_failure_stays_retryable_without_false_conflict():
+    result = run_node(
+        r"""
+const remote = {revision:7, device_id:'phone-b', identity:'issue:r:42:', elapsed_ms:9000, running:true};
+const calls = [];
+const statuses = [];
+const sync = createTodaySessionSync({
+  getDeviceId:()=> 'desktop-a',
+  fetchJson:async (url, options={}) => {
+    calls.push(options.method || 'GET');
+    if (!options.method) return remote;
+    const error = new Error('gateway unavailable');
+    error.status = 503;
+    throw error;
+  },
+  onStatus:status=>statuses.push(status),
+});
+(async()=>{
+  await sync.refresh();
+  const claimed = await sync.claim();
+  process.stdout.write(JSON.stringify({claimed, calls, statuses}));
+})().catch(error=>{console.error(error);process.exit(1);});
+"""
+    )
+
+    assert result["claimed"] is None
+    assert result["calls"] == ["GET", "PATCH"]
+    assert result["statuses"][-1] == "offline"
+    assert "conflict" not in result["statuses"]