Clear stale unanswered calls before new invite

server/calls.go

@@ -95,6 +95,8 @@ type videoCall struct {
 	callID string
 	// Product-level call type: audio or video.
 	callType string
+	// Time when the invite was accepted by the server.
+	startedAt time.Time
 	// Time when the call was accepted.
 	acceptedAt time.Time
 }
@@ -168,6 +170,13 @@ func (call *videoCall) matchesPayload(payload any) bool {
 	return payloadCallID == "" || payloadCallID == call.callID
 }
 
+func (call *videoCall) isStaleUnanswered(now time.Time, timeoutSeconds int) bool {
+	if call == nil || len(call.parties) != 1 || call.startedAt.IsZero() || timeoutSeconds <= 0 {
+		return false
+	}
+	return now.Sub(call.startedAt) > time.Duration(timeoutSeconds)*time.Second
+}
+
 func (t *Topic) auditCallCreate(call *videoCall, caller types.Uid) {
 	if store.Calls == nil || call == nil || call.callID == "" || caller.IsZero() {
 		return
@@ -350,6 +359,7 @@ func (t *Topic) handleCallInvite(msg *ClientComMessage, asUid types.Uid) {
 		contentMime: msg.Pub.Head["mime"],
 		callID:      callContentString(msg.Pub.Content, "callId"),
 		callType:    callContentString(msg.Pub.Content, "callType"),
+		startedAt:   types.TimeNow(),
 	}
 	t.currentCall.parties[msg.sess.sid] = callPartyData{
 		uid:          asUid,

server/datamodel.go

@@ -1634,6 +1634,18 @@ func ErrCallBusyReply(msg *ClientComMessage, ts time.Time) *ServerComMessage {
 	return ErrCallBusyExplicitTs(msg.Id, msg.Original, ts, msg.Timestamp)
 }
 
+func ErrCallBusyWithCallReply(msg *ClientComMessage, ts time.Time, call *videoCall) *ServerComMessage {
+	reply := ErrCallBusyReply(msg, ts)
+	if call != nil {
+		reply.Ctrl.Params = map[string]any{
+			"seq":       call.seq,
+			"call_id":   call.callID,
+			"call_type": call.callType,
+		}
+	}
+	return reply
+}
+
 // ErrUnknown database or other server error (500).
 func ErrUnknown(id, topic string, ts time.Time) *ServerComMessage {
 	return ErrUnknownExplicitTs(id, topic, ts, ts)

server/topic.go

@@ -1227,8 +1227,15 @@ func (t *Topic) handlePubBroadcast(msg *ClientComMessage) {
 			return
 		}
 		if t.currentCall != nil {
-			msg.sess.queueOut(ErrCallBusyReply(msg, types.TimeNow()))
+			now := types.TimeNow()
-			return
+			if t.currentCall.isStaleUnanswered(now, imCallTimeoutSeconds()) {
+				logs.Warn.Printf("topic[%s]: clearing stale unanswered call seq=%d call_id=%s before new invite", t.name, t.currentCall.seq, t.currentCall.callID)
+				t.terminateCallInProgress(true)
+			}
+			if t.currentCall != nil {
+				msg.sess.queueOut(ErrCallBusyWithCallReply(msg, now, t.currentCall))
+				return
+			}
 		}
 		head, content, err := normalizeCallInvite(msg.Pub.Head, msg.Pub.Content)
 		if err != nil {

server/topic_test.go

@@ -530,6 +530,28 @@ func TestHandleBroadcastCallRequiresStructuredInvite(t *testing.T) {
 	}
 }
 
+func TestVideoCallStaleUnanswered(t *testing.T) {
+	now := time.Date(2026, time.September, 4, 10, 5, 0, 0, time.UTC)
+	call := &videoCall{
+		parties:   map[string]callPartyData{"origin": {isOriginator: true}},
+		startedAt: now.Add(-31 * time.Second),
+	}
+	if !call.isStaleUnanswered(now, 30) {
+		t.Fatal("unanswered call older than establishment timeout should be stale")
+	}
+
+	call.startedAt = now.Add(-29 * time.Second)
+	if call.isStaleUnanswered(now, 30) {
+		t.Fatal("unanswered call within establishment timeout should not be stale")
+	}
+
+	call.parties["callee"] = callPartyData{}
+	call.startedAt = now.Add(-31 * time.Second)
+	if call.isStaleUnanswered(now, 30) {
+		t.Fatal("accepted call must not be treated as stale by establishment timeout")
+	}
+}
+
 func TestHandleBroadcastDataGroup(t *testing.T) {
 	topicName := "grp-test"
 	numUsers := 4