Merge pull request 'Make cross-device Today session sync conflict-convergent' (#1031) from timmy/1030-today-session-conflict-convergence into main

frontend/index.html

@@ -285,6 +285,7 @@
     <div class="small" id="my-work-action-status" aria-live="assertive"></div>
     <button class="retry-work-route" id="retry-unfiled-draft-sync" type="button" hidden>Retry Draft sync</button>
     <div class="small" id="today-sync-status" aria-live="polite">Today is saved on this device.</div>
+    <div class="small" id="today-session-sync-status" role="status" aria-live="polite" hidden></div>
     <div class="small" id="later-sync-status" aria-live="polite">Later is saved on this device.</div>
     <button class="retry-work-route" id="retry-work-route" type="button" hidden>Retry shared work item</button>
     <div class="small" id="work-route-share-status" aria-live="polite"></div>

frontend/today-session-sync.js

@@ -5,9 +5,19 @@ function createTodaySessionSync({
   let current = null;
   let ownedRevision = 0;
   let pollTimer = null;
+  let requestTail = Promise.resolve();
+  let latestSnapshot = null;
+  let publishRequested = 0;
+  let publishSent = 0;
   const endpoint = 'api/v1/today/session';
   const deviceId = () => String(getDeviceId?.() || '').trim();
 
+  function serialize(operation) {
+    const result = requestTail.then(operation, operation);
+    requestTail = result.catch(() => null);
+    return result;
+  }
+
   function adopt(session) {
     if (!session || !Number.isInteger(session.revision)) return null;
     const previousOwned = current?.device_id === deviceId() && current?.running;
@@ -27,8 +37,9 @@ function createTodaySessionSync({
     return session;
   }
 
-  async function refresh() {
+  async function refreshNow() {
     try {
+      onStatus('syncing');
       const session = await fetchJson(endpoint);
       onStatus('online');
       return adopt(session);
@@ -38,29 +49,39 @@ function createTodaySessionSync({
     }
   }
 
-  async function claim() {
+  const refresh = () => serialize(refreshNow);
-    if (!current?.running || !current.identity || current.device_id === deviceId() || !deviceId()) return null;
+
-    try {
+  function claim() {
-      const session = await fetchJson(endpoint, {
+    return serialize(async () => {
-        method:'PATCH', headers:{'Content-Type':'application/json'},
+      if (!current?.running || !current.identity || current.device_id === deviceId() || !deviceId()) return null;
-        body:JSON.stringify({
+      try {
-          base_revision:current.revision, device_id:deviceId(), identity:current.identity,
+        onStatus('syncing');
-          elapsed_ms:current.elapsed_ms, running:true,
+        const session = await fetchJson(endpoint, {
-        }),
+          method:'PATCH', headers:{'Content-Type':'application/json'},
-      });
+          body:JSON.stringify({
-      adopt(session);
+            base_revision:current.revision, device_id:deviceId(), identity:current.identity,
-      timer?.adopt?.(session.identity, session.elapsed_ms, session.running);
+            elapsed_ms:current.elapsed_ms, running:true,
-      return session;
+          }),
-    } catch (error) {
+        });
-      onStatus('conflict', error);
+        adopt(session);
-      await refresh();
+        timer?.adopt?.(session.identity, session.elapsed_ms, session.running);
-      return null;
+        return session;
-    }
+      } catch (error) {
+        if (error?.status === 409 || error?.code === 'session_changed') {
+          onStatus('conflict', error);
+          await refreshNow();
+        } else {
+          onStatus('offline', error);
+        }
+        return null;
+      }
+    });
   }
 
-  async function publish(snapshot = timer?.snapshot?.()) {
+  async function publishNow(snapshot) {
     if (!snapshot?.identity || !deviceId()) return null;
     try {
+      onStatus('syncing');
       const session = await fetchJson(endpoint, {
         method:'PATCH', headers:{'Content-Type':'application/json'},
         body:JSON.stringify({
@@ -72,11 +93,36 @@ function createTodaySessionSync({
       adopt(session);
       return session;
     } catch (error) {
-      onStatus('offline', error);
+      if (error?.status === 409 || error?.code === 'session_changed') {
+        onStatus('conflict', error);
+        await refreshNow();
+      } else {
+        onStatus('offline', error);
+      }
       return null;
     }
   }
 
+  function publish(snapshot = timer?.snapshot?.()) {
+    snapshot = snapshot?.identity ? snapshot : latestSnapshot;
+    if (!snapshot?.identity || !deviceId()) return Promise.resolve(null);
+    latestSnapshot = snapshot;
+    publishRequested += 1;
+    return serialize(async () => {
+      if (publishSent === publishRequested) return current;
+      if (current?.device_id && current.device_id !== deviceId()) {
+        publishSent = publishRequested;
+        return current;
+      }
+      const version = publishRequested;
+      const result = await publishNow(latestSnapshot);
+      publishSent = current?.device_id && current.device_id !== deviceId()
+        ? publishRequested
+        : version;
+      return result;
+    });
+  }
+
   const pulse = () => current?.device_id === deviceId() && current?.running
     ? publish()
     : refresh();
@@ -113,8 +159,17 @@ function attachTodaySessionHandoff({
     return value;
   };
   let offered = null;
+  const sessionStatus = qs('#today-session-sync-status');
+  const showSessionStatus = message => {
+    sessionStatus.textContent = message;
+    sessionStatus.hidden = !message;
+  };
   const sync = createTodaySessionSync({
     fetchJson, getDeviceId, timer,
+    onStatus:state => showSessionStatus(
+      state === 'syncing' ? 'Session syncing…' :
+        (state === 'offline' ? 'Session offline · will retry.' : '')
+    ),
     onRemote:session => {
       offered = session;
       const handoff = qs('#today-session-handoff');
@@ -126,6 +181,7 @@ function attachTodaySessionHandoff({
         `${item?.title || 'Current Today item'} · ${minutes} min elapsed`;
     },
     onTransferred:() => {
+      showSessionStatus('Session continued on another device.');
       announce('Today continued on another device. Timer paused here.');
       renderTimer();
     },

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"]