fix(social): complete shared listening request lifecycle

packages/core/src/listening.test.ts

@@ -31,4 +31,48 @@ describe('shared listening',()=>{
  it('pauses on connection loss and on room termination',async()=>{
   const e=setup();await e.coordinator.poll(true);e.get.mockRejectedValueOnce(new TypeError('offline'));e.advance(11000);await e.coordinator.poll(true);expect(e.adapter.changed).toHaveBeenLastCalledWith(expect.objectContaining({disconnected:true}));e.setState({...e.state,room:null});await e.coordinator.poll(true);expect(e.coordinator.locked).toBe(false);expect(e.adapter.changed).toHaveBeenLastCalledWith(expect.objectContaining({room:null}));e.coordinator.dispose();
  });
+ it('keeps a reserved source room alive without announcing a stream before approval',async()=>{
+  const e=setup(host,device);e.setState({...e.state,room:{...room,participants:[{id:host,name:'Host'}]},requests:[{id:'request',owner_id:host,sender_id:guest,sender_name:'Guest',recipient_id:host,kind:'join',expires_at:'2026-10-23T00:02:00Z'}]});
+  await e.coordinator.poll(true);
+  expect(e.adapter.changed).toHaveBeenLastCalledWith(expect.objectContaining({room:null,locked:false}));
+  expect(e.put).toHaveBeenCalled();expect(e.adapter.pause).not.toHaveBeenCalled();expect(e.adapter.apply).not.toHaveBeenCalled();e.coordinator.dispose();
+ });
+ it('releases an empty source reservation after its final request is cancelled or expires',async()=>{
+  const e=setup(host,device);e.setState({...e.state,room:{...room,participants:[{id:host,name:'Host'}]}});
+  await e.coordinator.poll(true);
+  expect(e.post).toHaveBeenCalledWith('/api/v1/me/social/listening/actions',expect.objectContaining({body:expect.objectContaining({action:'end',target_id:room.id})}));
+  expect(e.put).not.toHaveBeenCalled();expect(e.adapter.pause).not.toHaveBeenCalled();
+  expect(e.adapter.changed).toHaveBeenLastCalledWith(expect.objectContaining({room:null,locked:false}));
+  e.coordinator.dispose();expect(e.post).toHaveBeenCalledTimes(1);
+ });
+ it('does not take over another source device while its request is pending',async()=>{
+  const e=setup(host);e.setState({...e.state,room:{...room,participants:[{id:host,name:'Host'}]}});
+  await e.coordinator.poll(true);
+  expect(e.adapter.changed).toHaveBeenLastCalledWith(expect.objectContaining({room:null,locked:false}));
+  expect(e.adapter.apply).not.toHaveBeenCalled();expect(e.adapter.pause).not.toHaveBeenCalled();e.coordinator.dispose();expect(e.post).not.toHaveBeenCalled();
+ });
+ it('does not start a pending guest request with no admitted room',async()=>{
+  const e=setup();e.setState({...e.state,room:null,requests:[{id:'request',owner_id:host,sender_id:guest,sender_name:'Guest',recipient_id:host,kind:'join',expires_at:'2026-10-23T00:02:00Z'}]});
+  await e.coordinator.poll(true);expect(e.adapter.apply).not.toHaveBeenCalled();expect(e.coordinator.locked).toBe(false);e.coordinator.dispose();
+ });
+ it('requires explicit audio on a second owner device and does not end the source when it closes',async()=>{
+  const e=setup(host);await e.coordinator.poll(true);
+  expect(e.adapter.apply).toHaveBeenLastCalledWith(expect.objectContaining({playing:false}),3000,expect.any(Function));
+  e.coordinator.beginAudio();await vi.waitFor(()=>expect(e.adapter.apply).toHaveBeenLastCalledWith(expect.objectContaining({playing:true}),3000,expect.any(Function)));
+  e.coordinator.dispose();expect(e.post).not.toHaveBeenCalled();
+ });
+ it('discards a room lookup that arrives after a local exit',async()=>{
+  const e=setup();let resolve!:(value:{data:ListeningState;response:Response})=>void;
+  e.get.mockImplementationOnce(()=>new Promise(done=>{resolve=done;}));
+  const pending=e.coordinator.poll(true);e.coordinator.leaveLocally();resolve({data:e.state,response:new Response()});await pending;
+  expect(e.adapter.apply).not.toHaveBeenCalled();expect(e.adapter.changed).toHaveBeenLastCalledWith(expect.objectContaining({room:null}));e.coordinator.dispose();
+ });
+ it('queues a refresh requested while a state lookup is in flight',async()=>{
+  const e=setup();let resolve!:(value:{data:ListeningState;response:Response})=>void;
+  e.get.mockImplementationOnce(()=>new Promise(done=>{resolve=done;}));
+  const pending=e.coordinator.poll(true);e.coordinator.refresh();
+  e.setState({...e.state,room:null});resolve({data:{...e.state,room:{...room}},response:new Response()});await pending;
+  await vi.waitFor(()=>expect(e.adapter.changed).toHaveBeenLastCalledWith(expect.objectContaining({room:null})));
+  expect(e.get).toHaveBeenCalledTimes(2);e.coordinator.dispose();
+ });
 });

packages/core/src/listening.ts

@@ -22,6 +22,7 @@ export class ListeningCoordinator {
   private generation = 0;
   private sequence = 0;
   private busy = false;
+  private refreshPending = false;
   private closed = false;
   private lastPoll = 0;
   private lastSuccess = 0;
@@ -46,9 +47,12 @@ export class ListeningCoordinator {
     if (!room || !this.source(room)) return;
     const state = this.adapter.read();
     const key = `${state.track?.id}:${state.playing}`;
-    if (key !== this.key) { this.key = key; void this.poll(true); }
+    if (key !== this.key) { this.key = key; this.refresh(); }
+  }
+  refresh() {
+    if (this.busy) { this.refreshPending = true; return; }
+    void this.poll(true);
   }
-  refresh() { void this.poll(true); }
   allowJoining() { this.ignored.clear(); }
   beginAudio() { if (this.snapshot?.room) this.enabled.add(this.snapshot.room.id); this.refresh(); }
   leaveLocally() {
@@ -66,9 +70,10 @@ export class ListeningCoordinator {
     if (this.closed || this.busy || (!force && !this.snapshot?.room && this.now() - this.lastPoll < 10_000)) return;
     if (this.session.state?.status !== 'signed_in') return;
     this.busy = true; this.lastPoll = this.now();
+    const requestGeneration = this.generation;
     try {
       const state = unwrap(await this.session.client.GET('/api/v1/me/social/listening'));
-      if (this.closed) return;
+      if (this.closed || requestGeneration !== this.generation) return;
       this.lastSuccess = this.now();
       const previous = this.snapshot?.room;
       this.snapshot = state;
@@ -80,14 +85,32 @@ export class ListeningCoordinator {
         await this.session.client.POST('/api/v1/me/social/listening/actions', { body: { id: this.deviceId, action: 'leave', target_id: room.id } });
         return;
       }
+      // The server reserves a room while a request is pending. Only admitted
+      // participants turn it into a shared session; keep the source heartbeat
+      // alive without displaying a stream or taking over another device.
+      const shared = room.participants.some(person => person.id !== room.owner_id);
+      if (!shared && !this.source(room)) {
+        if (this.locked) { this.generation++; this.adapter.pause(); }
+        this.emit(null);
+        return;
+      }
       if (this.source(room)) {
-        this.emit(room);
+        // A cancelled or expired reservation must not keep an invisible room
+        // alive forever and prevent its owner from joining another stream.
+        if (!shared && !state.requests.some(request => request.owner_id === room.owner_id)) {
+          unwrap(await this.session.client.POST('/api/v1/me/social/listening/actions', { body: { id: this.deviceId, action: 'end', target_id: room.id } }));
+          this.snapshot = { ...state, room: null };
+          this.emit(null);
+          return;
+        }
+        this.emit(shared ? room : null);
         const playback = this.adapter.read();
         unwrap(await this.session.client.PUT('/api/v1/me/social/listening/state', { body: { room_id: room.id, device_id: this.deviceId, sequence: ++this.sequence, track_id: playback.track?.id, playing: playback.playing && !!playback.track, position_ms: Math.max(0, Math.round(playback.positionMs)) } }));
         return;
       }
       if (!this.locked) this.adapter.pause();
-      const needsStart = !state.preferences.autoplay && !this.enabled.has(room.id);
+      const ownOtherDevice = room.owner_id === this.session.state.user.id;
+      const needsStart = (!state.preferences.autoplay || ownOtherDevice) && !this.enabled.has(room.id);
       this.emit(room, needsStart);
       const version = this.generation;
       const elapsed = room.playing ? Math.max(0, Date.parse(state.server_time) - Date.parse(room.updated_at)) : 0;
@@ -95,7 +118,10 @@ export class ListeningCoordinator {
     } catch {
       if (this.closed) return;
       if (this.locked && this.now() - this.lastSuccess > 10_000) { this.adapter.pause(); this.emit(this.snapshot?.room ?? null, false, true); }
-    } finally { this.busy = false; }
+    } finally {
+      this.busy = false;
+      if (this.refreshPending && !this.closed) { this.refreshPending = false; this.refresh(); }
+    }
   }
   dispose() {
     if (this.closed) return;
@@ -104,7 +130,8 @@ export class ListeningCoordinator {
     clients.get(this.session)?.delete(this);
     const room = this.snapshot?.room;
     if (this.locked) this.adapter.pause();
-    if (room) void this.session.client.POST('/api/v1/me/social/listening/actions', { body: { id: this.deviceId, action: this.source(room) ? 'end' : 'leave', target_id: room.id } }).catch(() => undefined);
+    const ownAccount = this.session.state.status === 'signed_in' && room?.owner_id === this.session.state.user.id;
+    if (room && (this.source(room) || !ownAccount)) void this.session.client.POST('/api/v1/me/social/listening/actions', { body: { id: this.deviceId, action: this.source(room) ? 'end' : 'leave', target_id: room.id } }).catch(() => undefined);
   }
 }