package com.mentra.acsmeeting import android.content.Context import android.os.SystemClock import android.util.Base64 import android.util.Log import com.azure.android.communication.calling.AudioStreamBufferDuration import com.azure.android.communication.calling.AudioStreamChannelMode import com.azure.android.communication.calling.AudioStreamFormat import com.azure.android.communication.calling.AudioStreamSampleRate import com.azure.android.communication.calling.AudioStreamState import com.azure.android.communication.calling.AudioStreamType import com.azure.android.communication.calling.CapabilitiesCallFeature import com.azure.android.communication.calling.CapabilitiesChangedListener import com.azure.android.communication.calling.CommonCall import com.azure.android.communication.calling.CommonCallAgent import com.azure.android.communication.calling.TeamsCallAgent import com.azure.android.communication.calling.TeamsCallAgentOptions import com.azure.android.communication.calling.JoinTeamsCallOptions import com.azure.android.communication.calling.CallAgent import com.azure.android.communication.calling.CallAgentOptions import com.azure.android.communication.calling.CallClient import com.azure.android.communication.calling.CallState import com.azure.android.communication.calling.Features import com.azure.android.communication.calling.HangUpOptions import com.azure.android.communication.calling.ParticipantCapabilityType import com.azure.android.communication.calling.DiagnosticFlagChangedListener import com.azure.android.communication.calling.DiagnosticQualityChangedListener import com.azure.android.communication.calling.LocalUserDiagnosticsCallFeature import com.azure.android.communication.calling.MediaStatisticsCallFeature import com.azure.android.communication.calling.MediaStatisticsReportReceivedListener import com.azure.android.communication.calling.NetworkDiagnostics import com.azure.android.communication.calling.IncomingAudioOptions import com.azure.android.communication.calling.IncomingMixedAudioEvent import com.azure.android.communication.calling.JoinCallOptions import com.azure.android.communication.calling.LocalOutgoingAudioStream import com.azure.android.communication.calling.OutgoingAudioOptions import com.azure.android.communication.calling.OutgoingVideoConstraints import com.azure.android.communication.calling.OutgoingVideoOptions import com.azure.android.communication.calling.RawIncomingAudioStream import com.azure.android.communication.calling.RawIncomingAudioStreamOptions import com.azure.android.communication.calling.RawIncomingAudioStreamProperties import com.azure.android.communication.calling.RawOutgoingAudioStream import com.azure.android.communication.calling.RawOutgoingAudioStreamOptions import com.azure.android.communication.calling.RawOutgoingAudioStreamProperties import com.azure.android.communication.calling.RawOutgoingVideoStreamOptions import com.azure.android.communication.calling.TeamsMeetingLinkLocator import com.azure.android.communication.calling.VirtualOutgoingVideoStream import com.azure.android.communication.common.CommunicationTokenCredential import com.mentra.acsmeeting.audio.AcsAudioPolicy import com.mentra.acsmeeting.audio.ActiveStreamKind import com.mentra.acsmeeting.audio.JoinAudioPlan import com.mentra.acsmeeting.audio.AudioPolicyApplier import com.mentra.acsmeeting.audio.AudioSafety import com.mentra.acsmeeting.audio.AudioSourceKind import com.mentra.acsmeeting.audio.AudioStreamController import com.mentra.acsmeeting.audio.AudioUplinkChain import com.mentra.acsmeeting.audio.CallGuard import com.mentra.acsmeeting.audio.ExecutorPolicyScheduler import com.mentra.acsmeeting.audio.GlassesPcmRouting import com.mentra.acsmeeting.audio.AcsUplinkTransport import com.mentra.acsmeeting.audio.IncomingAudioPump import com.mentra.acsmeeting.audio.IncomingRateProbe import com.mentra.acsmeeting.audio.PcmBridge import com.mentra.acsmeeting.audio.PhoneMicCapturer import com.mentra.acsmeeting.audio.UplinkPacer import com.mentra.acsmeeting.audio.UplinkSender import com.mentra.acsmeeting.telemetry.CallDiagnostics import com.mentra.acsmeeting.telemetry.WireEpisodeTracker import com.mentra.glassesmedia.source.MediaDiagnostics import com.mentra.glassesmedia.source.CloudflareWhepSource import com.mentra.glassesmedia.source.DecoderMode import com.mentra.glassesmedia.source.OutgoingRateArm import com.mentra.glassesmedia.source.PixelFormatArm import com.mentra.glassesmedia.source.GlassesMediaController import com.mentra.glassesmedia.network.ScopedSoftApNetwork import com.mentra.glassesmedia.source.DecodedFrameTap import com.mentra.glassesmedia.source.GlassesMediaSourceFactory import com.mentra.glassesmedia.source.LocalWhipIngestSource import com.mentra.acsmeeting.source.MeetingVideoSourceSpec import com.mentra.glassesmedia.source.SourceConfig import com.mentra.glassesmedia.source.SourceKind import com.mentra.glassesmedia.source.SourceState import com.mentra.glassesmedia.source.SyntheticI420Source import com.mentra.glassesmedia.source.TargetSize import com.mentra.glassesmedia.source.VideoSourceArm import com.mentra.glassesmedia.telemetry.AvSyncProbe import com.mentra.glassesmedia.telemetry.PipelineStats import com.mentra.glassesmedia.telemetry.PipelineTicker import com.mentra.glassesmedia.trace.SoftApTrace import com.mentra.acsmeeting.video.AcsFrameSender import com.mentra.acsmeeting.video.VideoProfile import java.util.concurrent.CountDownLatch import java.util.concurrent.Executors import java.util.concurrent.Future import java.util.concurrent.ScheduledExecutorService import java.util.concurrent.ScheduledFuture import java.util.concurrent.TimeUnit import java.util.concurrent.TimeoutException import java.util.concurrent.atomic.AtomicReference import java.util.concurrent.atomic.AtomicBoolean import java.util.concurrent.atomic.AtomicInteger import kotlin.math.roundToInt class AcsMeetingSession( private val context: Context, private val onState: (Map) -> Unit, private val onIncomingPcm: (String, Int, Int) -> Unit, mediaSourceFactory: GlassesMediaSourceFactory? = null, /** * The joined glasses hotspot, when the call is a SoftAP call. Held so libwebrtc can be shown a * network Android hides from it; null for every Cloudflare call. */ private val scopedNetwork: ScopedSoftApNetwork? = null, ) { internal val stats = PipelineStats() private val avSync = AvSyncProbe() private val ticker = PipelineTicker(stats, avSync, onTick = { sampleBwe() }) { Log.i(TAG, it) } private val executor: ScheduledExecutorService = Executors.newSingleThreadScheduledExecutor() /** * How long a task sat on [executor] before it started running. * * Everything here is serialized onto one thread, so a `join` that appears to take 40 s may have * spent 38 of them queued behind the previous call's `leaveLocked`. Without this there is no way * to tell that apart from a slow ACS, and the two have opposite fixes. */ private fun traceQueued(stage: String, submittedAt: Long, vararg fields: Pair) { SoftApTrace.stage(stage, *fields, "queuedMs" to (SystemClock.elapsedRealtime() - submittedAt)) } private val scheduler = ExecutorPolicyScheduler(executor) private val outgoingReady = AtomicBoolean(false) private val muted = AtomicBoolean(false) private val frameSender = AcsFrameSender(stats, avSync) private var profile = VideoProfile.DEFAULT private val resolvedFactory = mediaSourceFactory ?: GlassesMediaSourceFactory { video, pcm, config -> // The synthetic diagnostic arm overrides everything; otherwise the requested kind decides. when { MediaDiagnostics.videoArm == VideoSourceArm.SYNTHETIC -> SyntheticI420Source(video, stats, frameSender::isReady) config.kind == SourceKind.SOFTAP -> LocalWhipIngestSource(context, video, pcm, stats, scopedNetwork) else -> CloudflareWhepSource(context, video, pcm, stats) } } @Volatile private var lastGatedLogMs = 0L private var pcmBridge: PcmBridge? = null /** * Ingest, delay, resample and pacing behind one lock, so mute means the same thing at every * stage. Built by [join] alongside the bridge it owns. */ @Volatile private var uplinkChain: AudioUplinkChain? = null /** * Whether the *host* is feeding this call's outgoing audio, rather than the decoded glasses * track. Set by the audio policy for a SoftAP call on the glasses microphone: the wearer's voice * arrives over BLE LC3 through [pushOutgoingPcm], and the WHIP peer publishes video only. * * Also a drop gate. Without it a host that pushes PCM at a call which is taking audio from the * media relay would put the same room on the call twice. */ private val externalPcmEnabled = AtomicBoolean(false) private val incomingProbe = IncomingRateProbe() // Clock-domain adapter: the WebRTC audio thread only ever fills the pacer, // and a dedicated monotonic-deadline thread drains it into ACS. private val pacer = UplinkPacer(log = { Log.i(TAG, it) }) @Volatile private var uplinkSender: UplinkSender? = null private val phoneMic = PhoneMicCapturer { pcm, rate, channels -> feedOutgoingPcm(pcm, rate, channels) } // Emits already-normalized 16 kHz mono; the host opens its PCM player with // exactly that format, so whatever ACS actually delivers cannot change pitch. private val incomingPump = IncomingAudioPump { chunk -> onIncomingPcm(PcmBridge.encodeBase64(chunk), IncomingAudioPump.OUT_RATE, IncomingAudioPump.OUT_CHANNELS) } // isSpeaking flips several times a second per participant; coalesce so the // host and miniapp see one roster snapshot per burst instead of a storm. private val rosterPushPending = AtomicBoolean(false) private val roster = RemoteRoster { if (rosterPushPending.compareAndSet(false, true)) { executor.schedule({ rosterPushPending.set(false) if (call != null) onState(snapshot()) }, ROSTER_COALESCE_MS, TimeUnit.MILLISECONDS) } } private val media = GlassesMediaController(resolvedFactory) private var mediaStatsListener: MediaStatisticsReportReceivedListener? = null private var mediaStatsFeature: MediaStatisticsCallFeature? = null private var capabilitiesFeature: CapabilitiesCallFeature? = null private var capabilitiesListener: CapabilitiesChangedListener? = null /** * Whether this participant may end the meeting for everyone. Teams grants it to presenters only, * and ACS can deliver it after admission, so it is a live value rather than a join-time fact. */ @Volatile private var hangUpForEveryone = CapabilityStatus() @Volatile private var manageLobby = CapabilityStatus() private val mediaStatsReports = AtomicInteger(0) /** Last wire size reported by ACS, so adaptation is logged on transition rather than every 1 Hz report. */ @Volatile private var lastWireSizeKey: String? = null private var netDiagnostics: NetworkDiagnostics? = null private var sendQualityListener: DiagnosticQualityChangedListener? = null private var reconnectListener: DiagnosticQualityChangedListener? = null private var noNetworkListener: DiagnosticFlagChangedListener? = null private var relaysListener: DiagnosticFlagChangedListener? = null private var callClient: CallClient? = null private var callAgent: CommonCallAgent? = null private var call: CommonCall? = null private var audioOut: RawOutgoingAudioStream? = null private var localOut: LocalOutgoingAudioStream? = null private var audioIn: RawIncomingAudioStream? = null private var videoOut: VirtualOutgoingVideoStream? = null /** Whether ACS is sending [videoOut]. Only [setVideoEnabled] turns it off; every call starts on. */ @Volatile private var videoEnabled = true @Volatile private var meetingUrl: String? = null @Volatile private var phase = "idle" @Volatile private var lastError: String? = null @Volatile private var audioSource = "glasses" @Volatile private var configuredAudioDelayMs = MediaDiagnostics.acsAudioDelayMs @Volatile private var lastSafety = AudioSafety.DEGRADED // Health of the glasses WHEP feed, reported alongside the ACS phase so the host // can tell "call is up, glasses video is dead" from a healthy call. @Volatile private var mediaSource = SourceState.IDLE // The transport of the active call. A SoftAP source must not be auto-rebuilt: its URL is an // output of binding, so a rebuild rebinds a new port and strands the glasses on the old one. @Volatile private var currentSourceKind = SourceKind.WHEP private var mediaRestartAttempts = 0 private var mediaRestartTask: ScheduledFuture<*>? = null /** * Which path the wearer took into this call: `created` (Start) or `joined` (Join). * * Stamped on every diagnostic below, because the open question these traces exist to answer — * whether a joined call runs at a lower bitrate than a created one — cannot be asked of a log * that does not say which kind of call produced each line. `unknown` means a caller that * predates the field, and is kept distinct from either answer rather than guessed at. */ @Volatile private var callOrigin = "unknown" /** ACS call id once the SDK has one; the only key that ties these lines to a Teams-side record. */ @Volatile private var callId = "" @Volatile private var joinStartedAtMs = 0L @Volatile private var lobbyEnteredAtMs = 0L @Volatile private var connectedAtMs = 0L @Volatile private var lastBweSampleAtMs = 0L /** * Calls out each stretch where the ACS uplink went bad, and what it cost to come back. * * Kept here rather than left to the analyzer because the two are not redundant: the analyzer can * only reconstruct episodes from readings that arrived, and on a Start call barely a quarter of * them do. This is the session's own account, so an episode is on the record even when * MEDIA_STATISTICS went quiet in the middle of it. */ @Volatile private var wireEpisodes = WireEpisodeTracker() /** When the wire size last changed, so the recovery ramp after a downscale is sampled densely. */ @Volatile private var lastAdaptationAtMs = 0L /** When the wire was last under the low threshold, for the same reason. */ @Volatile private var lastLowWireAtMs = 0L /** * Bumped by every join and every leave, so a bounded ACS operation that completes late cannot * attach its result to a session that has already moved on. * * Same pattern as the ingest source's generation. It exists because `createCallAgent` is a future * with no timeout of its own: on device it stalled for 30 s while cellular was still validating, * and a cancelled join that later succeeds would otherwise leave a live agent nobody owns. */ private val joinGeneration = AtomicInteger(0) /** * `createCallAgent` Future we stopped waiting on. ACS still finishes signing in; this is the only * handle that can dispose that leftover agent before the next join. */ private var abandonedAgent: Future? = null /** * Agent signed in before the glasses hotspot came up. SoftAP DNS cannot resolve ACS hosts, so * `createCallAgent` after the scoped join stalls until the hotspot is torn down. Join reuses this * instead of signing in again on the broken resolver. */ @Volatile private var agentPrepared = false private var preparedToken: String? = null private var preparedIdentityMode: String? = null private val controller = SessionAudioController() private val applier = AudioPolicyApplier(controller, scheduler) { Log.i(TAG, it) } fun snapshot(): Map { val result = mutableMapOf( "state" to phase, "muted" to muted.get(), "videoEnabled" to videoEnabled, "provider" to "acs-teams", "audioSource" to audioSource, "activeStream" to controller.readActive().name.lowercase(), "audioSafety" to lastSafety.name.lowercase(), "mediaSource" to mediaSource.name.lowercase(), "participants" to roster.snapshot(), // Nullable members inside, so the miniapp can tell "denied" from "not known yet" and only // offer End when it is actually allowed. "capabilities" to mapOf( "hangUpForEveryone" to hangUpForEveryone.toMap(), "manageLobby" to manageLobby.toMap(), ), ) meetingUrl?.let { result["meetingUrl"] = it } lastError?.let { result["error"] = it } media.ingestUrl?.let { result["ingestUrl"] = it } describeEndReason(call).forEach { (key, value) -> if (value != null) result["endReason_$key"] = value } return result } /** * Sign in to ACS before the glasses hotspot exists. * * On device, `createCallAgent` after the scoped join sat for the full 20 s deadline and only * completed once SoftAP was released. Token mint already proved the internet works *before* that * join. Doing this step on that same network is what makes the later SoftAP join a media bind * plus a Teams meeting join, not another sign-in through glasses dnsmasq. * * Waits [PREPARE_AGENT_WAIT_MS], not [CALL_AGENT_WAIT_MS]. The tight budget only ever existed * because sign-in used to run inside the SoftAP join, where overrunning cost the wearer the * hotspot too. Here nothing is torn down while we wait, and a cold sign-in on a slow AP measured * 35 s on device — under the old 20 s cap that agent arrived just in time to be thrown away. */ fun prepareAgent(token: String, displayName: String?, identityMode: String = "guest") { val done = CountDownLatch(1) val error = AtomicReference(null) phase = "connecting" lastError = null val submittedAt = SystemClock.elapsedRealtime() executor.execute { traceQueued("session_prepare_agent_begin", submittedAt) val startedAt = SystemClock.elapsedRealtime() try { leaveLocked(emitIdle = false) val generation = joinGeneration.get() lastError = null emit("connecting") val credential = CommunicationTokenCredential(token) callClient = CallClient() val agentOptions = CallAgentOptions() agentOptions.displayName = displayName ?: "Mentra Call" callAgent = obtainCallAgent(callClient!!, credential, agentOptions, generation, identityMode, PREPARE_AGENT_WAIT_MS) agentPrepared = true preparedToken = token preparedIdentityMode = identityMode Log.i(TAG, "ACS call agent prepared before SoftAP") SoftApTrace.stage( "session_prepare_agent_end", "durationMs" to (SystemClock.elapsedRealtime() - startedAt), ) } catch (failed: Exception) { agentPrepared = false lastError = formatAcsError(failed) Log.e(TAG, "prepare agent failed $lastError", failed) SoftApTrace.failure( "session_prepare_agent_failed", "durationMs" to (SystemClock.elapsedRealtime() - startedAt), "reason" to lastError, ) leaveLocked(emitIdle = false) emit("error") error.set(failed) } finally { done.countDown() } } if (!done.await(PREPARE_AGENT_WAIT_MS + ABANDONED_AGENT_REJOIN_WAIT_MS + 5_000L, TimeUnit.MILLISECONDS)) { SoftApTrace.failure( "session_prepare_agent_stuck", "waitedMs" to (SystemClock.elapsedRealtime() - submittedAt), ) throw IllegalStateException( "ACS_AGENT_TIMEOUT: Teams did not finish signing this phone in within " + "${PREPARE_AGENT_WAIT_MS / 1000}s.", ) } error.get()?.let { throw it } } fun join( token: String, teamsUrl: String, videoSource: MeetingVideoSourceSpec, displayName: String?, dumpWav: Boolean, audioSource: String = "glasses", video: VideoProfile = VideoProfile.DEFAULT, audioDelayMs: Int? = null, /** `created` (Start) or `joined` (Join); see [callOrigin]. Diagnostic only. */ origin: String = "unknown", identityMode: String = "guest", /** * Runs the WHIP listener bind, and exists so the caller can lift a process-wide network pin * across exactly that call. Takes the block rather than being a pair of before/after hooks so * the pin cannot be left off if the bind throws. See [network.InternetHold.bindProcessToCellular]. */ bindIngestUnpinned: (() -> Unit) -> Unit = { bind -> bind() }, ): Map { // Both glasses and phone feed RawOutgoingAudioStream so ACS never owns // the phone audio route (no MODE_IN_COMMUNICATION, no forced speaker). // Phone PCM comes from AudioRecord; glasses PCM still arrives via WHEP. val parsed = AcsAudioPolicy.parseSource(audioSource) ?: AudioSourceKind.GLASSES if (parsed == AudioSourceKind.PHONE) { Log.i(TAG, "audioSource=phone: AudioRecord → virtual outgoing; communication mode off") } this.audioSource = if (parsed == AudioSourceKind.PHONE) "phone" else "glasses" // The ACS work below is queued, so callers must not receive the pre-join // phase. Reflect the intent synchronously so the resolved snapshot is // "connecting" and cannot overwrite a fresher onState with a stale idle. phase = "connecting" lastError = null meetingUrl = teamsUrl callOrigin = origin callId = "" joinStartedAtMs = SystemClock.elapsedRealtime() lobbyEnteredAtMs = 0L connectedAtMs = 0L lastBweSampleAtMs = 0L // Fresh per call. A tracker carried over would open this call's first episode against the // previous call's floor, and an A/B run is back-to-back calls on different ceilings. wireEpisodes = WireEpisodeTracker() lastAdaptationAtMs = 0L lastLowWireAtMs = 0L // SoftAP needs the WHIP listener bound before this method returns: the JS // orchestrator reads ingestUrl off the join result and tears the scoped // network down if it is missing. ACS join itself stays on the executor. val softApReady = if (videoSource is MeetingVideoSourceSpec.SoftAp) CountDownLatch(1) else null val softApBindError = AtomicReference(null) val submittedAt = SystemClock.elapsedRealtime() executor.execute { traceQueued("session_join_begin", submittedAt, "transport" to videoSource.kind) val startedAt = SystemClock.elapsedRealtime() try { // Tear down any previous call without announcing idle: the caller already // holds a "connecting" snapshot, and an idle event landing after it made // the host and miniapp flash out of "joining" on every join. // A SoftAP join that already signed in on cellular must keep that agent: // recreating it on the glasses hotspot is the ACS_AGENT_TIMEOUT we just hit. val reuseAgent = agentPrepared && callAgent != null && preparedToken == token && preparedIdentityMode == identityMode leaveLocked(emitIdle = false, keepAgent = reuseAgent) // After the teardown, because that teardown bumps the generation itself. Everything that // moves it runs on this executor, so the value is stable for the rest of this join. val generation = joinGeneration.get() val requested = video this.profile = when (MediaDiagnostics.outgoingRate) { OutgoingRateArm.CLAMP_TO_SOFTWARE_CEILING -> requested.forSoftwareEncoder() OutgoingRateArm.ADVERTISE_REQUESTED -> requested } if (this.profile.fps != requested.fps) { Log.i( TAG, "ACS software-encoder clamp ${requested.width}x${requested.height}@${requested.fps} -> " + "@${this.profile.fps} (h264 sw holds ~${VideoProfile.SOFTWARE_ENCODER_FPS} fps; " + "advertising faster starves the wire)", ) } else { Log.i( TAG, "ACS rate arm=${MediaDiagnostics.outgoingRate} advertising " + "${this.profile.width}x${this.profile.height}@${this.profile.fps} unclamped; " + "P7 rate names what binds", ) } // Read by the 1 Hz verdict, which scores the wire against what we declared. stats.advertisedFps = this.profile.fps.toDouble() stats.budgetBps = this.profile.maxBitrateBps this.audioSource = if (parsed == AudioSourceKind.PHONE) "phone" else "glasses" meetingUrl = teamsUrl lastError = null emit("connecting") val bridge = PcmBridge(context.cacheDir, dumpWav) pcmBridge = bridge val delayMs = (audioDelayMs ?: MediaDiagnostics.acsAudioDelayMs) .coerceIn(0, AudioUplinkChain.MAX_DELAY_MS) configuredAudioDelayMs = delayMs val headroom = if (videoSource.kind == SourceKind.SOFTAP && parsed != AudioSourceKind.PHONE) { UplinkPacer.LC3_TARGET_MS } else UplinkPacer.TARGET_MS pacer.configureTarget(headroom) avSync.reset() uplinkChain = AudioUplinkChain( bridge, pacer, // Move delay into a jitter buffer that can absorb BLE bursts, rather than adding latency. (delayMs - (headroom - UplinkPacer.TARGET_MS)).coerceAtLeast(0), onIngest = { pcm, nowNs -> avSync.onAudio(pcm, nowNs) }, ) if (reuseAgent) { Log.i(TAG, "reusing call agent prepared before SoftAP") agentPrepared = false } else { val credential = CommunicationTokenCredential(token) callClient = CallClient() val agentOptions = CallAgentOptions() agentOptions.displayName = displayName ?: "Mentra Call" callAgent = obtainCallAgent(callClient!!, credential, agentOptions, generation, identityMode) } val videoOptions = RawOutgoingVideoStreamOptions() videoOptions.formats = listOf(AcsFrameSender.outgoingFormat(profile)) val videoStream = VirtualOutgoingVideoStream(videoOptions) videoOut = videoStream frameSender.attach( videoStream, onFormat = { size -> media.setTargetSize(size) }, // Fires at stream start and again on every renegotiation. ACS trades resolution inside // the budget it was given, so "what we asked for" and "what is being sent" diverge // silently; only this says when. onNegotiatedFormat = { format -> SoftApTrace.stage( "acs_outgoing_format", *diag( "width" to format.width, "height" to format.height, "fps" to format.framesPerSecond, "pixelFormat" to format.pixelFormat, "askedWidth" to profile.width, "askedHeight" to profile.height, "askedFps" to profile.fps, "maxBitrateBps" to profile.maxBitrateBps, "rateArm" to MediaDiagnostics.outgoingRate.name.lowercase(), ), ) }, ) val audioProperties = RawOutgoingAudioStreamProperties() .setFormat(AudioStreamFormat.PCM16_BIT) .setSampleRate(AudioStreamSampleRate.HZ_48000) .setChannelMode(AudioStreamChannelMode.MONO) .setBufferDuration(AudioStreamBufferDuration.MS20) val outAudioOptions = RawOutgoingAudioStreamOptions().setProperties(audioProperties) val outgoing = RawOutgoingAudioStream(outAudioOptions) audioOut = outgoing outgoing.addOnStateChangedListener { val ready = outgoing.state.toString().contains("STARTED", ignoreCase = true) outgoingReady.set(ready) Log.i(TAG, "raw outgoing audio state=${outgoing.state}") if (ready) startUplink(outgoing) else stopUplink() applyAudioPolicy("virtual-stream-state") } val synthetic = MediaDiagnostics.videoArm == VideoSourceArm.SYNTHETIC if (synthetic) muted.set(true) val desired = desiredKind() val plan = if (synthetic) { JoinAudioPlan(armVirtual = true, transportMuted = true) } else { AcsAudioPolicy.planJoin(desired, muted.get(), GLASSES_REQUIRES_UNMUTED_TRANSPORT) } // Virtual outgoing stays armed for phone and glasses. A LocalOutgoing // stream would make ACS own the phone route and open an echo loop. val local = if (plan.armVirtual) null else LocalOutgoingAudioStream() localOut = local local?.addOnStateChangedListener { Log.i(TAG, "local outgoing audio state=${local.state}") applyAudioPolicy("local-stream-state") } val incomingProperties = RawIncomingAudioStreamProperties() .setFormat(AudioStreamFormat.PCM16_BIT) .setSampleRate(AudioStreamSampleRate.HZ_16000) .setChannelMode(AudioStreamChannelMode.MONO) val inAudioOptions = RawIncomingAudioStreamOptions().setProperties(incomingProperties) val incoming = RawIncomingAudioStream(inAudioOptions) audioIn = incoming incomingPump.reset() incomingProbe.reset() incoming.addOnStateChangedListener { Log.i(TAG, "raw incoming audio state=${incoming.state}") } incoming.addOnMixedAudioBufferReceivedListener { event: IncomingMixedAudioEvent -> if (MediaDiagnostics.videoArm == VideoSourceArm.SYNTHETIC) return@addOnMixedAudioBufferReceivedListener try { val data = event.audioBuffer?.buffer ?: return@addOnMixedAudioBufferReceivedListener val bytes = ByteArray(data.remaining()) data.get(bytes) // Trust the event's format over what we asked for. val props = event.streamProperties val rate = sampleRateHz(props?.sampleRate) ?: 16000 val channels = if (props?.channelMode == AudioStreamChannelMode.STEREO) 2 else 1 incomingPump.push(bytes, rate, channels) logIncomingRate(rate, channels, bytes.size) } catch (error: Exception) { Log.w(TAG, "incoming PCM callback failed", error) } } val joinOptions = JoinCallOptions() val constraints = OutgoingVideoConstraints() .setMaxWidth(profile.width) .setMaxHeight(profile.height) .setMaxFrameRate(profile.fps) .setMaxBitrateInBps(profile.maxBitrateBps) val ov = OutgoingVideoOptions() .setOutgoingVideoStreams(listOf(videoStream)) .setConstraints(constraints) joinOptions.setOutgoingVideoOptions(ov) // Virtual outgoing and MODE_IN_COMMUNICATION off for glasses and phone. // Communication mode makes the ACS SDK call setMode(3) on connect and // request the phone speaker, which yanks A2DP off the glasses. val oa = OutgoingAudioOptions() .setStream(if (plan.armVirtual) outgoing else requireNotNull(local)) .setMuted(plan.transportMuted) .setCommunicationAudioModeEnabled(!plan.armVirtual) joinOptions.setOutgoingAudioOptions(oa) // Raw incoming replaces SDK playout: buffers come to us, the SDK plays // nothing. Do not start it "speaker muted" — that flag gates the very // stream we read from. val ia = IncomingAudioOptions() .setStream(incoming) .setMuted(false) joinOptions.setIncomingAudioOptions(ia) stats.arm = when { synthetic -> "synthetic" videoSource is MeetingVideoSourceSpec.SoftAp -> "softap" else -> "whep" } stats.pathMode = if (MediaDiagnostics.decoderMode == DecoderMode.BYTE_BUFFER) "bytebuf" else "texture" stats.pathCopy = when { MediaDiagnostics.pixelFormat == PixelFormatArm.NV12 -> "nv12" MediaDiagnostics.zeroCopy -> "zerocopy" else -> "planes" } stats.pix = MediaDiagnostics.pixelFormat.name.lowercase() stats.zcOn = if (MediaDiagnostics.zeroCopy) 1 else 0 mediaRestartAttempts = 0 currentSourceKind = if (synthetic) SourceKind.DIRECT else videoSource.kind media.setStateListener { state, reason -> // Fired from WebRTC/OkHttp threads; hop to the session executor so it // serializes with join/leave/policy like everything else. executor.execute { onMediaSourceState(state, reason) } } // SoftAP: bind the listener before the Teams join so the Expo promise can // return an ingest URL while 192.168.43.x is still assigned. WHEP still // attaches after the call exists — it has a URL going in, not coming out. if (videoSource is MeetingVideoSourceSpec.SoftAp) { // Unpinned for the bind only: this socket is a ServerSocket on 192.168.43.79 and cannot // be re-scoped afterwards, so a cellular mark here leaves the glasses' SYN unanswered. // The pin is back on by the time the Teams join below runs. bindIngestUnpinned { media.attach( video = { planes -> // Preview first, deliberately: the tap must see frames ACS's pacing and // readiness gates would otherwise hide, and it cannot delay or break this call. DecodedFrameTap.offer(planes) if (frameSender.sendPlanes(planes)) DecodedFrameTap.recordAcsSend() }, pcm = { pcm, rate, channels -> feedOutgoingPcm(pcm, rate, channels) }, config = videoSource.toConfig(), ) } } val locator = TeamsMeetingLinkLocator(teamsUrl) val joined = when (val agent = callAgent!!) { is TeamsCallAgent -> agent.join(context, locator, JoinTeamsCallOptions() .setOutgoingVideoOptions(ov).setOutgoingAudioOptions(oa).setIncomingAudioOptions(ia)) is CallAgent -> agent.join(context, locator, joinOptions) else -> throw IllegalStateException("Unsupported meeting agent") } call = joined roster.attach(joined) joined.addOnStateChangedListener { pushCallState(joined.state) } joined.addOnOutgoingAudioStateChangedListener { Log.i(TAG, "outgoing audio state changed muted=${joined.isOutgoingAudioMuted}") applyAudioPolicy("outgoing-audio-state") } pushCallState(joined.state) if (videoSource !is MeetingVideoSourceSpec.SoftAp) { media.attach( video = { planes -> DecodedFrameTap.offer(planes) if (frameSender.sendPlanes(planes)) DecodedFrameTap.recordAcsSend() }, pcm = { pcm, rate, channels -> feedOutgoingPcm(pcm, rate, channels) }, config = if (synthetic) SourceConfig("", SourceKind.DIRECT) else videoSource.toConfig(), ) } media.setTargetSize(TargetSize(profile.width, profile.height)) ticker.start() Log.i( TAG, "CPU probe: 1 Hz P6 ladder includes path{mode copy}, buf{tex i420}, stride{tight padded}, " + "copyP95, zc{}, cpu{proc}. i420P95 is toI420; copyP95 is the single plane copy. " + "BYTE_BUFFER should drop i420P95 and set buf{tex=0}. zerocopy should drop copyP95.", ) applyAudioPolicy("join") Log.i( TAG, "ACS join started arm=${MediaDiagnostics.videoArm.name.lowercase()} " + "profile=${profile.width}x${profile.height}@${profile.fps} " + "maxBitrate=${profile.maxBitrateBps} bitsPerFrame=${profile.bitsPerFrame()} " + "syntheticFps=${MediaDiagnostics.syntheticFps} entropy=${MediaDiagnostics.syntheticEntropy} " + "decoderMode=${MediaDiagnostics.decoderMode} zeroCopy=${MediaDiagnostics.zeroCopy} " + "pixelFormat=${MediaDiagnostics.pixelFormat} " + "source=${this.audioSource} audio=${if (synthetic) "off" else "on"} " + "armVirtual=${plan.armVirtual} transportMuted=${plan.transportMuted}", ) SoftApTrace.stage( "session_join_end", "durationMs" to (SystemClock.elapsedRealtime() - startedAt), "reusedAgent" to reuseAgent, "ingestUrl" to (media.ingestUrl ?: "none"), ) if (generation != joinGeneration.get()) { throw IllegalStateException("The meeting was cancelled before it finished joining") } softApReady?.countDown() } catch (error: Exception) { val message = formatAcsError(error) Log.e(TAG, "join failed $message", error) SoftApTrace.failure( "session_join_failed", "durationMs" to (SystemClock.elapsedRealtime() - startedAt), "reason" to message, "ingestUrl" to (media.ingestUrl ?: "none"), ) // A step after a successful ACS join (e.g. WHEP start) can throw. Record // the failure before tearing the call down: lastError makes pushCallState // ignore the hang-up's async disconnected callbacks, and emitIdle=false // keeps the terminal state as error instead of resetting to idle. Either // would otherwise let Mentra Call treat the failed join as a clean end. lastError = message softApBindError.compareAndSet(null, error) leaveLocked(emitIdle = false) emit("error") softApReady?.countDown() } } if (softApReady != null) { if (!softApReady.await(SOFTAP_JOIN_WAIT_MS, TimeUnit.MILLISECONDS)) { // The executor never got far enough to bind the listener. It is still running, so this // failure leaves work in flight that the next join's barrier has to wait out. SoftApTrace.failure( "session_join_bind_timeout", "waitedMs" to (SystemClock.elapsedRealtime() - submittedAt), "timeoutMs" to SOFTAP_JOIN_WAIT_MS, ) throw IllegalStateException("SoftAP ingest listener did not bind in ${SOFTAP_JOIN_WAIT_MS}ms") } softApBindError.get()?.let { throw it } if (media.ingestUrl == null) { throw IllegalStateException("SoftAP ingest listener bound but produced no URL") } } return snapshot() } /** * Point the subscriber at a different WHEP URL. WHEP only: a SoftAP listener has no URL to * update, since the URL is an output of binding, and its recovery is a full rebuild through * [restartVideoSource]. */ fun updateVideoSource(whepUrl: String) { executor.execute { // The host has a fresher opinion about where the glasses publish; drop // any automatic retry against the old URL. cancelMediaRestart() media.restart(SourceConfig(whepUrl)) } } /** * The URL the glasses must POST their offer to, for a SoftAP call. Null until the listener has * bound, and null for every other transport. The orchestrator reads this after join and sends it * to the glasses in `start_stream`. */ fun softApIngestUrl(): String? = media.ingestUrl /** * Destroy the current SoftAP listener generation and bind a new one on the address the * scoped network reports *now* — the caller rejoins first, then asks for this. * * Blocks on [executor] so it cannot race [join] / [leaveLocked]. Returns the new ingest URL * the glasses must be told; throws rather than hand back a stale listener. */ fun rebindSoftApIngest(bindIngestUnpinned: (() -> Unit) -> Unit = { bind -> bind() }): String { val done = CountDownLatch(1) val result = AtomicReference(null) val failure = AtomicReference(null) val submittedAt = SystemClock.elapsedRealtime() executor.execute { try { traceQueued("session_rebind_ingest_begin", submittedAt) if (!MediaDiagnostics.SOFTAP_RECOVERY_ENABLED) { throw IllegalStateException("SoftAP ingest rebind is disabled") } if (currentSourceKind != SourceKind.SOFTAP) { throw IllegalStateException("rebindSoftApIngest is only valid for a SoftAP call") } val address = scopedNetwork?.localIpv4() ?: throw IllegalStateException("scoped network has no IPv4 address after rejoin") // Same pin-lift as [join]: a listener bound while this UID is marked cellular // never receives the glasses' TCP SYN, even though ping/ARP succeed. lateinit var url: String bindIngestUnpinned { url = media.rebindIngest( SourceConfig("", SourceKind.SOFTAP, address), REBIND_INGEST_CLOSE_MS, ) } SoftApTrace.stage("session_rebind_ingest_end", "url" to url) result.set(url) } catch (error: Exception) { SoftApTrace.failure( "session_rebind_ingest_failed", "reason" to "${error.javaClass.simpleName}: ${error.message ?: ""}", ) failure.set(error) } finally { done.countDown() } } if (!done.await(REBIND_INGEST_CLOSE_MS + 8_000L, TimeUnit.MILLISECONDS)) { SoftApTrace.failure("session_rebind_ingest_timeout", "timeoutMs" to (REBIND_INGEST_CLOSE_MS + 8_000L)) throw IllegalStateException("SoftAP ingest rebind timed out") } failure.get()?.let { throw it } return result.get() ?: throw IllegalStateException("SoftAP ingest rebind returned no URL") } /** * Rebuild the WHEP subscription on the current URL even when it looks healthy. * The host calls this when the phone changed networks: ICE may not have noticed * yet, but the old candidate pair is dead. */ fun restartVideoSource() { executor.execute { // A SoftAP source cannot be rebuilt in place: forceRestart() rebinds a new OS-chosen port and // mints a fresh ingestUrl, but the glasses keep POSTing to the old port and nothing re-pushes // the new URL, which permanently strands the call in CONNECTING. Leaving the listener bound on // its stable port instead lets the glasses' own WHIP reconnect recover against the same URL; // a network-level recovery is the orchestrator's job (re-run the SoftapCallTransport sequence). if (currentSourceKind == SourceKind.SOFTAP) { Log.w(TAG, "restartVideoSource ignored for SoftAP; a rebuild would strand the glasses on a dead port") return@execute } cancelMediaRestart() media.forceRestart() } } private fun onMediaSourceState(state: SourceState, reason: String?) { val previous = mediaSource mediaSource = state if (state == SourceState.LIVE) mediaRestartAttempts = 0 if (state == SourceState.FAILED) scheduleMediaRestart(reason) // start() emits IDLE then CONNECTING back to back; one snapshot per real change. if (previous != state && call != null && phase != "idle") onState(snapshot()) } /** * Native owns first-line recovery: nothing above this layer can see ICE fail, and * a Teams call with a frozen last frame looks healthy from every other angle. * Exponential backoff capped at MEDIA_RESTART_MAX_MS; runs as long as the call is * alive. The host resets it whenever it hands us a new URL. */ private fun scheduleMediaRestart(reason: String?) { if (call == null || phase == "idle" || phase == "disconnected" || phase == "error") return // SoftAP recovery must never rebuild the source from here: forceRestart() rebinds a new port and // ingestUrl the glasses are never told about, so the call would strand in CONNECTING. The // FAILED state is still surfaced to the host (onMediaSourceState emits a snapshot), and the // still-bound listener lets the glasses' own WHIP reconnect recover on the unchanged URL. if (currentSourceKind == SourceKind.SOFTAP) { Log.w(TAG, "SoftAP media source failed ($reason); session-level rebuild suppressed, listener left bound for glasses reconnect") return } if (mediaRestartTask?.isDone == false) return val attempt = mediaRestartAttempts++ val delayMs = minOf(MEDIA_RESTART_BASE_MS shl minOf(attempt, 4), MEDIA_RESTART_MAX_MS) Log.w(TAG, "glasses media source failed ($reason); WHEP rebuild #${attempt + 1} in ${delayMs}ms") mediaRestartTask = executor.schedule({ mediaRestartTask = null if (call == null || mediaSource != SourceState.FAILED) return@schedule try { media.forceRestart() } catch (error: Exception) { Log.w(TAG, "WHEP rebuild failed", error) scheduleMediaRestart("rebuild_threw") } }, delayMs, TimeUnit.MILLISECONDS) } private fun cancelMediaRestart() { mediaRestartTask?.cancel(false) mediaRestartTask = null mediaRestartAttempts = 0 } fun setMuted(next: Boolean): Map { if (MediaDiagnostics.videoArm == VideoSourceArm.SYNTHETIC) { muted.set(true) return snapshot() } muted.set(next) // Before the executor hop, not after it. Muting is the one audio operation whose latency the // wearer can hear as a mistake, and the policy queue can be several ACS round trips deep. if (next) uplinkChain?.mute() else uplinkChain?.unmute() executor.execute { applyAudioPolicy("set-muted") } val snap = snapshot() onState(snap) return snap } /** * Stop or resume the camera Teams receives without leaving the call. The glasses source, the * preview tap and the WHEP/WHIP transport keep running; [frameSender] drops frames while ACS * reports the stream stopped. [videoEnabled] only moves once ACS accepts the change. */ fun setVideoEnabled(next: Boolean, complete: (Map?, Throwable?) -> Unit) { val generation = joinGeneration.get() executor.execute { val active = call val stream = videoOut if (active == null || stream == null || joinGeneration.get() != generation) { complete(null, IllegalStateException("No active meeting")) return@execute } if (videoEnabled == next) { complete(snapshot(), null) return@execute } val result = runCatching { val change = if (next) active.startVideo(context, stream) else active.stopVideo(context, stream) change.get(VIDEO_TOGGLE_TIMEOUT_MS, TimeUnit.MILLISECONDS) } if (call !== active || joinGeneration.get() != generation) { complete(null, IllegalStateException("The meeting ended before the camera changed")) return@execute } result.exceptionOrNull()?.let { error -> Log.e(TAG, "setVideoEnabled=$next failed", error) complete(null, error) return@execute } Log.i(TAG, "setVideoEnabled=$next") videoEnabled = next val snap = snapshot() onState(snap) complete(snap, null) } } fun setAudioSource(source: String): Map { val parsed = AcsAudioPolicy.parseSource(source) if (parsed == null) { Log.w(TAG, "unknown audioSource=$source ignored; source is locked for this call") return snapshot() } Log.i(TAG, "setAudioSource=$source ignored; audio source is locked for this call at ${audioSource}") return snapshot() } fun leave() { val submittedAt = SystemClock.elapsedRealtime() // Invalidate before queueing: leaveLocked cannot run until the executor finishes the current // join/hang-up, and without this bump a Cancel sits behind an unbounded ACS Future. joinGeneration.incrementAndGet() // Queued and unwaited, so the only evidence this leave ever ran is the line the task logs // when it starts. A `leave` with no matching begin means the executor never reached it. SoftApTrace.stage("session_leave_queued") executor.execute { traceQueued("session_leave_begin", submittedAt) leaveLocked() } } /** * Leave, and do not return until the cleanup has actually finished. * * [leave] hands the work to [executor] and returns straight away, so a caller that awaits it and * then starts the next call is racing this one's hang-up, agent disposal, and media teardown * through the same hardware. That race is what turned a quick Stop/Start into a hotspot the * previous call switched off underneath the new one. * * Failures that [leaveLocked] otherwise only logs are captured and rethrown here: reporting a * clean teardown that did not happen is exactly what lets the next call build on leaked state. * * @param timeoutMs how long the cleanup may take before it is reported as stuck * @throws IllegalStateException when the cleanup did not finish in time */ fun leaveAndAwait(timeoutMs: Long): Boolean { val done = CountDownLatch(1) val failure = AtomicReference(null) val submittedAt = SystemClock.elapsedRealtime() joinGeneration.incrementAndGet() executor.execute { traceQueued("session_leave_and_await_begin", submittedAt) val startedAt = SystemClock.elapsedRealtime() try { leaveLocked(failures = failure) SoftApTrace.stage( "session_leave_and_await_end", "durationMs" to (SystemClock.elapsedRealtime() - startedAt), // Recorded rather than only logged: this is the one the host rethrows and turns into a // refusal for the next call. "recordedFailure" to (failure.get()?.let { "${it.javaClass.simpleName}: ${it.message ?: ""}" } ?: "none"), ) } catch (error: Exception) { SoftApTrace.failure( "session_leave_and_await_failed", "durationMs" to (SystemClock.elapsedRealtime() - startedAt), "reason" to "${error.javaClass.simpleName}: ${error.message ?: ""}", ) failure.compareAndSet(null, error) } finally { done.countDown() } } if (!done.await(timeoutMs, TimeUnit.MILLISECONDS)) { // The cleanup is still running. Nothing is abandoned and nothing is safe to restart yet, // which is why this is reported rather than treated as a finished teardown. SoftApTrace.failure("session_leave_and_await_timeout", "timeoutMs" to timeoutMs) throw IllegalStateException("acs_leave_timeout") } failure.get()?.let { throw it } return true } /** * Wait for the SoftAP WHIP listener to release its port, and say whether it did. * * Deliberately not folded into [leaveAndAwait]: the ACS teardown and the listener's tombstone * run on their own clocks, and collapsing them into one answer would hide which of the two the * next call is actually waiting on. `false` means the port is still held. */ fun awaitIngestClosed(timeoutMs: Long): Boolean = media.awaitIngestClosed(timeoutMs) /** Drop the retiring WHIP listener now. The barrier's forced path, after [awaitIngestClosed]. */ fun forceCloseIngest() = media.forceCloseIngest() /** * End the Teams group call for everyone, then tear this device down. * * Blocking, unlike [leave], because the caller has to know whether the meeting actually died: the * miniapp shows different terminal copy for "ended" and "you left, but the meeting may still be * active", and inventing the first would be a lie the wearer cannot check. * * Local teardown is queued whatever the hang-up did. A refused or failed End must still get the * wearer out of the call — the only thing at stake in the failure is what we claim happened. * * @throws IllegalStateException when there is no call, when the capability is known to be denied, * or when ACS rejects the hang-up */ fun endForEveryone(): Map { val done = CountDownLatch(1) val failure = AtomicReference(null) val submittedAt = SystemClock.elapsedRealtime() executor.execute { traceQueued("session_end_for_everyone_begin", submittedAt) val startedAt = SystemClock.elapsedRealtime() try { val active = call ?: throw IllegalStateException("no_active_call") // Re-read rather than trusting the cached value: capabilities arrive asynchronously and the // last event may predate admission. val capability = readHangUpForEveryone() EndForEveryonePolicy.refusalFor(capability)?.let { throw IllegalStateException(it) } Log.i(TAG, "end for everyone: hangUp(forEveryone=true) allowed=${capability.allowed}") active.hangUp(HangUpOptions().setForEveryone(true)).get() Log.i(TAG, "end for everyone: ACS accepted the hang-up") SoftApTrace.stage( "session_end_for_everyone_end", "durationMs" to (SystemClock.elapsedRealtime() - startedAt), ) } catch (error: Exception) { Log.w(TAG, "end for everyone failed", error) SoftApTrace.failure( "session_end_for_everyone_failed", "durationMs" to (SystemClock.elapsedRealtime() - startedAt), "reason" to "${error.javaClass.simpleName}: ${error.message ?: ""}", ) failure.set(error) } finally { done.countDown() } } val settled = done.await(END_FOR_EVERYONE_WAIT_MS, TimeUnit.MILLISECONDS) if (!settled) { SoftApTrace.failure("session_end_for_everyone_timeout", "timeoutMs" to END_FOR_EVERYONE_WAIT_MS) } // Queued unconditionally, and after the await so it cannot dispose the agent out from under the // hang-up. A timed-out End still leaves this device. executor.execute { leaveLocked() } if (!settled) throw IllegalStateException("end_for_everyone_timeout") failure.get()?.let { throw it } return snapshot() } /** Latest capability read, for a host that wants to enable or hide End before the user taps it. */ fun hangUpForEveryoneCapability(): CapabilityStatus = hangUpForEveryone fun getState(): Map = snapshot() /** The generation guard also rejects a result arriving after Leave was requested. */ fun admitParticipant(participantId: String, complete: (Throwable?) -> Unit) { val generation = joinGeneration.get() executor.execute { val active = call if (active == null) { complete(IllegalStateException("No connected meeting")) return@execute } admitLobbyParticipant( active, participantId, readCapability(ParticipantCapabilityType.MANAGE_LOBBY).allowed == true, isCurrent = { call === active && joinGeneration.get() == generation }, dispatch = { executor.execute(it) }, complete = { error -> if (error == null) onState(snapshot()) else Log.e(TAG, "Selected lobby participant admission failed", error) complete(error) }, ) } } private fun desiredKind(): AudioSourceKind = if (audioSource == "phone") AudioSourceKind.PHONE else AudioSourceKind.GLASSES /** Queues onto [executor] so ACS callbacks cannot race the applier. */ private fun applyAudioPolicy(reason: String) { executor.execute { applyAudioPolicyOnExecutor(reason) } } private fun applyAudioPolicyOnExecutor(reason: String) { lastSafety = applier.apply(desiredKind(), muted.get(), reason) if (lastSafety == AudioSafety.UNSAFE) { Log.e(TAG, "audioSafety=unsafe — mute and stopAudio both failed; unintended mic may be live") } onState(snapshot()) } private fun logIncomingRate(rate: Int, channels: Int, bytes: Int) { val reading = incomingProbe.record(System.nanoTime(), bytes, rate, channels) ?: return Log.i( TAG, "P8 audio-in declaredRate=${reading.declaredRate} ch=${reading.channels} " + "samplesPerCallback=${reading.samplesPerCallback} " + "callbackHz=${"%.2f".format(reading.callbackHz)} " + "measuredRate=${reading.measuredRate.roundToInt()} " + "prerollMs=${IncomingAudioPump.DEFAULT_PREROLL_MS} events=${incomingPump.eventsIn} " + "in=${incomingPump.bytesIn} out16k=${incomingPump.bytesOut} " + "formatChanges=${incomingPump.formatChanges}", ) } private fun feedOutgoingPcm(pcm: ByteArray, sampleRate: Int, channels: Int) { if (muted.get()) { val now = System.currentTimeMillis() if (now - lastGatedLogMs >= 1000) { lastGatedLogMs = now Log.i(TAG, "outgoing PCM gated bytes=${pcm.size} (user muted)") } return } if (!outgoingReady.get()) return // Resample here, but do not touch ACS: sending straight from this thread // hands ACS the glasses' audio clock in bursts. The pacer decides when. uplinkChain?.ingest(pcm, sampleRate, channels) } /** * Accept one buffer of microphone PCM from the host. * * The BLE LC3 path: the glasses encode their microphone, the Bluetooth SDK decodes it on the * phone, and the host forwards it here rather than the wearer's voice riding the WHIP track. It * is deliberately a hard drop rather than a buffer when there is no call to feed — a frame kept * for a session that is going away is a frame played into the *next* call. * * @return true when the buffer entered the uplink */ fun pushOutgoingPcm(base64: String, sampleRate: Int, channels: Int): Boolean { if (!externalPcmEnabled.get()) return false if (phase == "idle" || phase == "disconnected" || phase == "error") return false val pcm = try { Base64.decode(base64, Base64.DEFAULT) } catch (error: IllegalArgumentException) { Log.w(TAG, "pushOutgoingPcm got undecodable base64 len=${base64.length}", error) return false } if (pcm.isEmpty()) return false feedOutgoingPcm(pcm, sampleRate, channels) return true } @Synchronized private fun startUplink(stream: RawOutgoingAudioStream) { if (uplinkSender != null) return pacer.reset() val sender = UplinkSender( pacer, AcsUplinkTransport(stream), muted = { muted.get() }, pcmMeanAbs = { pcmBridge?.lastMeanAbs ?: -1 }, ) uplinkSender = sender sender.start() // The A/V configuration this call ran with, stated once at the top so a receiver recording can // be attributed to it. The measured ingest offset is a separate `AVSYNC clap audioLeadMs` // line. Never print the two as one number: a delay that is configured is not an offset that // was observed, and conflating them is how a calibration gets believed. Log.i( TAG, "P8 audio-up config configuredDelayMs=$configuredAudioDelayMs " + "audioTimestamps=${MediaDiagnostics.acsAudioTimestamps}", ) } @Synchronized private fun stopUplink() { uplinkSender?.stop() uplinkSender = null pacer.reset() } private fun pushCallState(state: CallState) { // A failed join has already reported a terminal error and torn the call // down; ignore any late ACS state callback so it cannot overwrite error. if (lastError != null) return val previous = phase phase = when (state) { CallState.CONNECTING -> "connecting" CallState.IN_LOBBY -> "lobby" CallState.CONNECTED -> "connected" CallState.DISCONNECTING -> phase CallState.DISCONNECTED -> "disconnected" else -> phase } val end = describeEndReason(call) // A failed connection is recoverable; do not tell the wearer the remote meeting ended. // Wait for DISCONNECTED because DISCONNECTING may not yet carry the final reason. if (state == CallState.DISCONNECTED && (end["code"] as? Number)?.toInt()?.let { it != 0 } == true) { phase = "error" lastError = "ACS_CONNECTION_LOST: The Teams connection was lost. Check mobile data and rejoin. " + "(code=${end["code"]}, subcode=${end["subcode"]})" } Log.i(TAG, "ACS call state=$state phase=$phase previous=$previous end=$end") if (callId.isEmpty()) callId = readCallId() if (phase == "lobby" && lobbyEnteredAtMs == 0L) lobbyEnteredAtMs = SystemClock.elapsedRealtime() if (phase == "connected" && connectedAtMs == 0L) connectedAtMs = SystemClock.elapsedRealtime() SoftApTrace.stage( "acs_call_state", *diag( "state" to state.toString().lowercase(), // Not `phase`: the trace parser treats a bare `phase=` as a stage name, because that is // how the host's own lines are shaped. A field called `phase` here is silently dropped. "callPhase" to phase, "previous" to previous, "sinceJoinMs" to sinceJoinMs(), // Reported on every transition, not only on admission, so a call that is still waiting // shows a dwell that grows instead of a field that appears once at the end. "sinceLobbyMs" to lobbyDwellMs(), "endCode" to (end["code"] ?: -1), "endSubcode" to (end["subcode"] ?: -1), ), ) if (phase == "connected" && previous != "connected") { call?.let { attachMediaStats(it) attachDiagnostics(it) attachCapabilities(it) } applyAudioPolicy("call-connected") } else { onState(snapshot()) } } private fun emit(next: String) { phase = next onState(snapshot()) } /** The ACS call id, or empty if the SDK has not minted one yet. Never throws into a trace. */ private fun readCallId(): String = try { call?.id.orEmpty() } catch (error: Exception) { Log.w(TAG, "call id unavailable", error) "" } private fun attachMediaStats(joined: CommonCall) { detachMediaStats() mediaStatsReports.set(0) lastWireSizeKey = null try { val feature = joined.feature(Features.MEDIA_STATISTICS) val listener = MediaStatisticsReportReceivedListener { event -> val outgoing = event.report?.outgoingStatistics val videos = outgoing?.videoStatistics val n = mediaStatsReports.incrementAndGet() val video = videos?.firstOrNull() if (n <= 8 || video == null) { Log.i( TAG, "P6 wire report #$n videos=${videos?.size ?: 0} " + "audios=${outgoing?.audioStatistics?.size ?: 0} " + "codec=${video?.codecName ?: "na"} fps=${video?.frameRate ?: "na"}", ) } stats.wireFps = video?.frameRate?.toDouble() stats.wireWidth = video?.frameWidth stats.wireHeight = video?.frameHeight stats.wireBitrateBps = video?.bitrateInBps?.toLong() stats.wirePacketCount = video?.packetCount val codec = video?.codecName.orEmpty() if (codec.isNotBlank() && codec != stats.codecName) { Log.i(TAG, "P6 wire codec=$codec ${video?.frameWidth}x${video?.frameHeight} fps=${video?.frameRate}") } stats.codecName = codec reportWireAdaptation(video?.frameWidth, video?.frameHeight, video?.frameRate) val width = video?.frameWidth val height = video?.frameHeight if (width != null && height != null && width > 0 && height > 0) { stats.setSize(width, height) } // Every report, not just the first eight: Start has been observed to attach and then // never print a P6 line the analyzer can see. SoftApTrace is what the comparison reads. SoftApTrace.stage( "acs_media_stats", *CallDiagnostics.mediaStatsReportFields( origin = callOrigin, callId = callId, n = n, videos = videos?.size ?: 0, audios = outgoing?.audioStatistics?.size ?: 0, wireBitrateBps = video?.bitrateInBps?.toLong(), width = video?.frameWidth, height = video?.frameHeight, fps = video?.frameRate?.toDouble(), codec = codec, packetCount = video?.packetCount, ), ) } feature.addOnReportReceivedListener(listener) mediaStatsListener = listener mediaStatsFeature = feature // First attempt often throws while ACS is still spinning up the media // stack (S26 Ultra never emitted a default-interval report). Retry after // CONNECTED so codecName is not stuck at na. scheduleMediaStatsInterval(feature, 0) Log.i(TAG, "P6 wire hop attached") SoftApTrace.stage("acs_media_stats_attach", *CallDiagnostics.mediaStatsAttachFields(callOrigin, callId, ok = true)) } catch (error: Exception) { Log.w(TAG, "MEDIA_STATISTICS attach failed", error) SoftApTrace.stage( "acs_media_stats_attach", *CallDiagnostics.mediaStatsAttachFields( callOrigin, callId, ok = false, error = "${error.javaClass.simpleName}:${error.message.orEmpty()}", ), ) } } /** * Logs a change in what ACS is actually putting on the wire versus the profile we asked for. * * ACS runs its own rate controller and will trade resolution for frames inside the budget it was * given, so the negotiated 1280x720 is a ceiling, not a promise. Without this the ladder prints * the adapted size once per second and a permanent downscale reads exactly like a healthy call — * the number is right there and nothing ever calls it out. Logged on transition only, because at * 1 Hz a warning per report is noise nobody reads. * * Traced as well as logged. A `Log.w` is invisible to `acs-quality-compare.mjs`, and while that * was the only record, a call that spent 90 s at 320x180 was scored as healthy: the analyzer had * no downscale to count, and 320x180 at a steady 15 fps looks like a working ladder. The trace * carries `ceilingBps` so a downscale can be attributed to its A/B arm. */ private fun reportWireAdaptation(width: Int?, height: Int?, fps: Float?) { if (width == null || height == null || width <= 0 || height <= 0) return val key = "${width}x$height" val previous = lastWireSizeKey if (key == previous) return lastWireSizeKey = key val askedPixels = profile.width.toLong() * profile.height val gotPixels = width.toLong() * height // First report is the baseline, not an adaptation: direction is only meaningful against a // previous size, and calling the initial negotiated size a "downscale" would put a spurious // adaptation on every call in the A/B. val direction = when { previous == null -> "initial" gotPixels < pixelsOfKey(previous) -> "down" else -> "up" } lastAdaptationAtMs = SystemClock.elapsedRealtime() SoftApTrace.stage( "acs_wire_adaptation", *CallDiagnostics.wireAdaptationFields( origin = callOrigin, callId = callId, width = width, height = height, direction = direction, sinceConnectedMs = if (connectedAtMs == 0L) -1L else SystemClock.elapsedRealtime() - connectedAtMs, askedWidth = profile.width, askedHeight = profile.height, wireBitrateBps = stats.wireBitrateBps, ceilingBps = profile.maxBitrateBps, fps = fps?.toDouble(), ), ) if (gotPixels < askedPixels) { val percent = (gotPixels * 100 / askedPixels).toInt() Log.w( TAG, "P6 wire ADAPTED_DOWN acs=$key (${percent}% of ${profile.width}x${profile.height}) " + "fps=${fps ?: "na"} codec=${stats.codecName.ifBlank { "na" }} " + "budget=${profile.maxBitrateBps} bitsPerFrame=${profile.bitsPerFrame()}", ) } else { Log.i(TAG, "P6 wire size=$key at or above profile ${profile.width}x${profile.height}") } } private fun pixelsOfKey(key: String): Long { val parts = key.split("x") val width = parts.getOrNull(0)?.toLongOrNull() ?: return 0L val height = parts.getOrNull(1)?.toLongOrNull() ?: return 0L return width * height } /** * Keep asking ACS for 1 Hz reports until it agrees, for as long as the call lasts. * * This used to give up after six attempts across about ten seconds, and the captures show it * never once succeeded: every call in `outputs/acs-quality` carries `interval=no`, and the Start * calls sit at 0-28% observed coverage as a direct result. The default interval is coarse enough * that a 90-second collapse can pass with a handful of readings, so losing this is what made the * whole investigation guess. * * Ten seconds was the wrong budget because it encodes the wrong theory. The evidence is that ACS * populates outgoing statistics only once a remote subscriber is actually pulling the stream — * which on a Start call is whenever the other person happens to join, not a fixed delay after * the local join. So the retry now backs off and keeps going for * [MEDIA_STATS_INTERVAL_MAX_ATTEMPTS], and stops early only on success or when the feature is * replaced. */ private fun scheduleMediaStatsInterval(feature: MediaStatisticsCallFeature, attempt: Int) { val delaySeconds = when { attempt == 0 -> 0L attempt <= 5 -> 2L attempt <= 15 -> 5L else -> 15L } executor.schedule({ if (mediaStatsFeature !== feature) return@schedule try { feature.updateReportIntervalInSeconds(1) Log.i(TAG, "P6 wire interval=1s attempt=$attempt") SoftApTrace.stage( "acs_media_stats_interval", *CallDiagnostics.mediaStatsIntervalFields( callOrigin, callId, attempt = attempt, ok = true, seconds = 1, ), ) } catch (error: Exception) { Log.w( TAG, "P6 wire interval attempt=$attempt failed ${error.javaClass.simpleName}: ${error.message}", ) SoftApTrace.stage( "acs_media_stats_interval", *CallDiagnostics.mediaStatsIntervalFields( callOrigin, callId, attempt = attempt, ok = false, seconds = 1, error = "${error.javaClass.simpleName}:${error.message.orEmpty()}", ), ) // Logged at info for the first handful and then only occasionally: the point of retrying // for minutes is defeated if it fills logcat with the same refusal every 15 seconds. if (attempt < MEDIA_STATS_INTERVAL_MAX_ATTEMPTS) { scheduleMediaStatsInterval(feature, attempt + 1) } else { Log.w( TAG, "P6 wire interval never accepted after $attempt attempts; " + "MEDIA_STATISTICS stays at its default cadence and captures will be sparse", ) } } }, delaySeconds, TimeUnit.SECONDS) } /** * OutgoingVideoStatistics reports frameRate/bitrate/packetCount and nothing * about loss or RTT, so `wire` cannot see a bad uplink. Microsoft's own * "sender's video is frozen" guidance points at these diagnostics instead: * they are the only send-side network signal the SDK exposes. */ private fun attachDiagnostics(joined: CommonCall) { detachDiagnostics() try { val feature = joined.feature(Features.LOCAL_USER_DIAGNOSTICS) as LocalUserDiagnosticsCallFeature val network = feature.networkDiagnostics val onSend = DiagnosticQualityChangedListener { args -> val quality = args.value?.name ?: "null" // Latched so `P9 quality` can reprint it every tick; this fires on change only. stats.sendQuality = quality logDiagnostic("networkSendQuality", quality) } val onReconnect = DiagnosticQualityChangedListener { args -> logDiagnostic("networkReconnectionQuality", args.value?.name ?: "null") } val onNoNetwork = DiagnosticFlagChangedListener { args -> logDiagnostic("networkUnavailable", args.value.toString()) } val onRelays = DiagnosticFlagChangedListener { args -> logDiagnostic("networkRelaysUnreachable", args.value.toString()) } network.addOnNetworkSendQualityChangedListener(onSend) network.addOnNetworkReconnectionQualityChangedListener(onReconnect) network.addOnIsNetworkUnavailableChangedListener(onNoNetwork) network.addOnIsNetworkRelaysUnreachableChangedListener(onRelays) netDiagnostics = network sendQualityListener = onSend reconnectListener = onReconnect noNetworkListener = onNoNetwork relaysListener = onRelays Log.i(TAG, "P7 diagnostics attached") } catch (error: Exception) { Log.w(TAG, "LOCAL_USER_DIAGNOSTICS attach failed", error) } } private fun logDiagnostic(name: String, value: String) { Log.i(TAG, "P7 diag $name=$value") SoftApTrace.stage("acs_network_diag", *diag("name" to name, "value" to value)) } /** * Stamp a diagnostic with the two facts that make it comparable across runs. * * Without `origin` the Start-versus-Join question cannot be asked of the log at all, and without * `callId` two calls in one capture merge into one timeline whenever the trace id is reused. */ private fun diag(vararg fields: Pair): Array> = CallDiagnostics.stamp(callOrigin, callId, *fields) private fun sinceJoinMs(): Long = if (joinStartedAtMs == 0L) -1L else SystemClock.elapsedRealtime() - joinStartedAtMs /** * How long this call has been in, or was held in, the Teams lobby. * * Frozen at admission rather than reset, so every sample after `connected` still carries the * dwell that preceded it. That is the correlation the comparison is looking for: a joined call * that settles low *and* waited in the lobby points at ACS rate control settling while there was * nowhere to send to. */ private fun lobbyDwellMs(): Long = when { lobbyEnteredAtMs == 0L -> -1L connectedAtMs > 0L -> connectedAtMs - lobbyEnteredAtMs else -> SystemClock.elapsedRealtime() - lobbyEnteredAtMs } /** * One `acs_bwe_sample` per cadence slot, driven by the 1 Hz ladder tick. * * The cadence follows the wire's health rather than the clock. It used to be dense (2 s) for the * first 90 s after `connected` and sparse (10 s) forever after, on the theory that a call which * settles low settles low early. An 8-minute capture killed that theory: the collapse started at * t=150 s, so the 90 seconds that mattered got nine samples while the uneventful first minute * got forty-five. Now anything other than an established full-rate call is worth 2 s — a dip, * the climb out of one, a fresh downscale, or ACS publishing empty reports. * * Before `connected` the dense rate applies too: the outgoing stream exists during the lobby, * and what it does there is half the hypothesis. */ private fun sampleBwe() { val now = SystemClock.elapsedRealtime() val sinceConnectedMs = if (connectedAtMs == 0L) -1L else now - connectedAtMs val wireBitrateBps = stats.wireBitrateBps if (wireBitrateBps != null && wireBitrateBps in 1 until CallDiagnostics.LOW_BITRATE_BPS) { lastLowWireAtMs = now } val health = CallDiagnostics.wireHealth( sinceConnectedMs = sinceConnectedMs, wireBitrateBps = wireBitrateBps, msSinceLow = if (lastLowWireAtMs == 0L) -1L else now - lastLowWireAtMs, msSinceAdaptation = if (lastAdaptationAtMs == 0L) -1L else now - lastAdaptationAtMs, ) // Episodes are tracked on every tick, not on every emitted sample: the record of a collapse // must not depend on the cadence that the collapse is what widens. trackWireEpisode(sinceConnectedMs, wireBitrateBps) val interval = CallDiagnostics.sampleIntervalMs(health) if (lastBweSampleAtMs != 0L && now - lastBweSampleAtMs < interval) return lastBweSampleAtMs = now SoftApTrace.stage( "acs_bwe_sample", *CallDiagnostics.bweFields( callOrigin, callId, CallDiagnostics.BweSample( state = phase, sinceJoinMs = sinceJoinMs(), sinceConnectedMs = sinceConnectedMs, lobbyDwellMs = lobbyDwellMs(), sendQuality = stats.sendQuality, wireBitrateBps = stats.wireBitrateBps, wireWidth = stats.wireWidth, wireHeight = stats.wireHeight, sentFps = stats.lastSubFps, wireFps = stats.wireFps, inboundBitrateBps = stats.inboundBitrateBps, inboundFps = stats.recvFps, decodedFps = stats.decodedFps, framesGated = stats.dropCount(), pacerDrops = stats.dropPacedCount(), budgetBps = stats.budgetBps, rateArm = MediaDiagnostics.outgoingRate.name.lowercase(), mediaStatsReports = mediaStatsReports.get(), mediaStatsAttached = mediaStatsFeature != null, videoOut = videoOutLabel(), packetsPerSecond = stats.lastPacketsPerSecond, subCount = stats.subCount(), sinkCount = stats.sinkCount(), ), ), ) } /** * Feed the episode tracker and emit the transitions it reports. * * Separate from the sample above because the two have different jobs. The sample answers "what * is it doing now" and is rate-limited; this answers "what did that outage cost" and must not * be, because the interesting transitions are exactly two per outage and dropping either one * loses the measurement. Only counted once connected — the lobby has nowhere to send to, so a * low rate there is not an outage. */ private fun trackWireEpisode(sinceConnectedMs: Long, wireBitrateBps: Long?) { if (sinceConnectedMs < 0) return val event = wireEpisodes.observe( atMs = sinceConnectedMs, bitrateBps = wireBitrateBps, width = stats.wireWidth, height = stats.wireHeight, inboundBitrateBps = stats.inboundBitrateBps?.toDouble(), sentFps = stats.lastSubFps, ) ?: return val label = event.phase.name.lowercase() SoftApTrace.stage( "acs_wire_episode", *CallDiagnostics.episodeFields( origin = callOrigin, callId = callId, episode = label, startedAtMs = event.startedAtMs, durationMs = event.durationMs, minBitrateBps = event.minBitrateBps, minResolution = event.minResolution, recoveryMs = event.recoveryMs, inboundBitrateBps = event.inboundBitrateBps, sentFps = event.sentFps, ceilingBps = profile.maxBitrateBps, ), ) // Warned rather than logged at info: a wire under 500 kbps while the glasses keep feeding full // rate is the fault itself, and it is the line a reader scanning logcat should trip over. Log.w( TAG, "P6 wire episode=$label since=${event.startedAtMs}ms dur=${event.durationMs}ms " + "floor=${event.minBitrateBps} minRes=${event.minResolution.ifBlank { "na" }} " + "recovery=${event.recoveryMs}ms glassesHop=${event.inboundBitrateBps?.toLong() ?: -1} " + "sentFps=${event.sentFps} ceiling=${profile.maxBitrateBps}", ) } /** ACS stream state, or `none` if we never built one. Independent of MEDIA_STATISTICS. */ private fun videoOutLabel(): String { val stream = videoOut ?: return "none" return try { stream.state.toString().lowercase() } catch (_: Exception) { "unknown" } } /** * Subscribe to participant capabilities so End can be offered honestly. * * The capability that matters is `HANG_UP_FOR_EVERYONE`. It can flip mid-call — a presenter role * granted or removed — so the listener stays attached rather than reading once at connect. */ private fun attachCapabilities(joined: CommonCall) { detachCapabilities() try { val feature = joined.feature(Features.CAPABILITIES) val listener = CapabilitiesChangedListener { event -> val changed = event.changedCapabilities.orEmpty().any { it.type == ParticipantCapabilityType.HANG_UP_FOR_EVERYONE || it.type == ParticipantCapabilityType.MANAGE_LOBBY } if (!changed) return@CapabilitiesChangedListener executor.execute { if (call !== joined || capabilitiesFeature !== feature) return@execute hangUpForEveryone = readHangUpForEveryone(feature) manageLobby = readCapability(ParticipantCapabilityType.MANAGE_LOBBY, feature) onState(snapshot()) } } feature.addOnCapabilitiesChangedListener(listener) capabilitiesFeature = feature capabilitiesListener = listener hangUpForEveryone = readHangUpForEveryone(feature) manageLobby = readCapability(ParticipantCapabilityType.MANAGE_LOBBY, feature) Log.i( TAG, "capability hangUpForEveryone allowed=${hangUpForEveryone.allowed} reason=${hangUpForEveryone.reason}", ) } catch (error: Exception) { // Unknown, not denied: an End is still attempted and ACS gets to answer. Log.w(TAG, "CAPABILITIES attach failed", error) hangUpForEveryone = CapabilityStatus(reason = "capabilities_unavailable") manageLobby = CapabilityStatus(reason = "capabilities_unavailable") } } private fun readHangUpForEveryone( feature: CapabilitiesCallFeature? = capabilitiesFeature, ): CapabilityStatus = readCapability(ParticipantCapabilityType.HANG_UP_FOR_EVERYONE, feature) private fun readCapability( type: ParticipantCapabilityType, feature: CapabilitiesCallFeature? = capabilitiesFeature, ): CapabilityStatus { val current = feature ?: return CapabilityStatus(reason = "capabilities_unavailable") return try { val capability = current.capabilities.orEmpty() .firstOrNull { it.type == type } ?: return CapabilityStatus(reason = "not_reported") CapabilityStatus(capability.isAllowed, capability.reason?.name?.lowercase()) } catch (error: Exception) { Log.w(TAG, "capabilities read failed", error) CapabilityStatus(reason = "capabilities_unavailable") } } private fun detachCapabilities() { val feature = capabilitiesFeature val listener = capabilitiesListener capabilitiesFeature = null capabilitiesListener = null if (feature == null || listener == null) return try { feature.removeOnCapabilitiesChangedListener(listener) } catch (_: Exception) { } } private fun detachDiagnostics() { val network = netDiagnostics ?: return try { sendQualityListener?.let { network.removeOnNetworkSendQualityChangedListener(it) } reconnectListener?.let { network.removeOnNetworkReconnectionQualityChangedListener(it) } noNetworkListener?.let { network.removeOnIsNetworkUnavailableChangedListener(it) } relaysListener?.let { network.removeOnIsNetworkRelaysUnreachableChangedListener(it) } } catch (_: Exception) { } netDiagnostics = null sendQualityListener = null reconnectListener = null noNetworkListener = null relaysListener = null } private fun detachMediaStats() { val listener = mediaStatsListener ?: return val feature = mediaStatsFeature mediaStatsListener = null mediaStatsFeature = null try { // Must be the same feature instance we added to: call.feature() can hand back a // fresh wrapper, and removing from that leaves the listener live on a leaving call. feature?.removeOnReportReceivedListener(listener) } catch (_: Exception) { } } /** * Sign in to ACS with a deadline, and make sure nothing survives a missed one. * * `createCallAgent` returns a future with no timeout of its own, and it is the first thing after * the hotspot join that needs the internet. Unbounded, a cellular route that had not validated * yet turned into a 30 s stall followed by the join step's own timeout — one opaque failure * covering a specific, nameable cause. * * Do not cancel that future. Cancel marks it done without a value, so the sweeper can no longer * recover the native agent ACS still creates. The next join then dies with "CallAgent associated * with this identity already exists". */ private fun obtainCallAgent( client: CallClient, credential: CommunicationTokenCredential, options: CallAgentOptions, generation: Int, identityMode: String, waitMs: Long = CALL_AGENT_WAIT_MS, ): CommonCallAgent { disposeAbandonedAgent(waitMs = ABANDONED_AGENT_REJOIN_WAIT_MS) return try { awaitCallAgent(client, credential, options, generation, identityMode, waitMs) } catch (error: Exception) { if (!AbandonedCallAgent.isExistingAgentError(error)) throw error Log.w(TAG, "createCallAgent hit leftover identity; disposing abandoned agent and retrying") disposeAbandonedAgent(waitMs = ABANDONED_AGENT_REJOIN_WAIT_MS) awaitCallAgent(client, credential, options, generation, identityMode, waitMs) } } private fun awaitCallAgent( client: CallClient, credential: CommunicationTokenCredential, options: CallAgentOptions, generation: Int, identityMode: String, waitMs: Long, ): CommonCallAgent { val startedAt = SystemClock.elapsedRealtime() SoftApTrace.stage("session_call_agent_wait", "waitMs" to waitMs, "generation" to generation) val pending: Future = when (identityMode) { "guest" -> client.createCallAgent(context, credential, options) "teams-user" -> client.createTeamsCallAgent(context, credential, TeamsCallAgentOptions()) else -> throw IllegalArgumentException("Unsupported meeting identity mode") } val agent = try { pending.get(waitMs, TimeUnit.MILLISECONDS) } catch (timeout: TimeoutException) { abandonedAgent = pending sweepLateAgent(pending, 0) // The sign-in is still running and still owns the identity. Named here because the next // join's "identity already exists" failure is otherwise the first sign of this one. SoftApTrace.failure( "session_call_agent_abandoned", "waitedMs" to (SystemClock.elapsedRealtime() - startedAt), "waitMs" to waitMs, ) throw IllegalStateException( "ACS_AGENT_TIMEOUT: Teams did not finish signing this phone in within " + "${waitMs / 1000}s. This step needs the internet, so it usually means mobile " + "data had not taken over yet after joining the glasses hotspot.", timeout, ) } // Leave now bumps generation off this executor, so a Cancel during `createCallAgent` can // land while we are blocked above. That is the case this check exists for. if (generation != joinGeneration.get()) { runCatching { agent.dispose() } SoftApTrace.failure( "session_call_agent_stale", "waitedMs" to (SystemClock.elapsedRealtime() - startedAt), "generation" to generation, "current" to joinGeneration.get(), ) throw IllegalStateException("ACS_AGENT_STALE: the call was torn down before Teams signed in") } SoftApTrace.stage("session_call_agent_ready", "waitedMs" to (SystemClock.elapsedRealtime() - startedAt)) return agent } /** * Dispose an agent that turns up after its wait was abandoned. * * Polls rather than chaining a completion callback because the SDK hands back a bare [Future]. * Gives up after a bounded number of sweeps: by then the process has either got the agent or the * future is never completing, and an endless timer is its own leak. */ private fun sweepLateAgent(pending: Future, sweep: Int) { if (sweep >= LATE_AGENT_SWEEPS) { Log.w(TAG, "abandoned call agent never completed; stopping sweep") // Giving up on the sweep is giving up on disposing that agent. It is a bounded leak by // design, but it is a leak, and the next sign-in is where it will be felt. SoftApTrace.failure("session_late_agent_abandoned", "sweeps" to sweep) return } executor.schedule({ if (abandonedAgent !== pending) return@schedule val late = AbandonedCallAgent.takeIfDone(pending) if (late == null) { if (!pending.isDone) sweepLateAgent(pending, sweep + 1) return@schedule } abandonedAgent = null Log.w(TAG, "disposing call agent that arrived after its join was abandoned") SoftApTrace.stage("session_late_agent_disposed", "sweep" to sweep) runCatching { late.dispose() } }, LATE_AGENT_SWEEP_MS, TimeUnit.MILLISECONDS) } /** * Best-effort dispose of a leftover agent before the next `createCallAgent`. * * [waitMs] is for rejoin: the previous sign-in may still be finishing, and that is the only * handle that can free the identity ACS refuses to share. */ private fun disposeAbandonedAgent(waitMs: Long) { val pending = abandonedAgent ?: return val startedAt = SystemClock.elapsedRealtime() val late = AbandonedCallAgent.take(pending, waitMs) if (late != null) { abandonedAgent = null Log.w(TAG, "disposing abandoned call agent before the next join") SoftApTrace.stage( "session_abandoned_agent_disposed", "waitedMs" to (SystemClock.elapsedRealtime() - startedAt), ) runCatching { late.dispose() } return } // Still not finished. The identity stays taken, so the `createCallAgent` about to run may be // refused — which is the retry [obtainCallAgent] exists for, not an unexplained join failure. SoftApTrace.failure( "session_abandoned_agent_unclaimed", "waitedMs" to (SystemClock.elapsedRealtime() - startedAt), "done" to pending.isDone, ) if (pending.isDone) abandonedAgent = null } /** * @param failures when present, the first cleanup exception is recorded here instead of only * being logged, so [leaveAndAwait] can tell its caller the teardown did not really succeed */ private fun leaveLocked( emitIdle: Boolean = true, keepAgent: Boolean = false, failures: AtomicReference? = null, ) { // Invalidate first: a bounded ACS operation still in flight has to find a stale generation // rather than attach an agent to a session that is being torn down. joinGeneration.incrementAndGet() val startedAt = SystemClock.elapsedRealtime() SoftApTrace.stage( "session_cleanup_begin", "emitIdle" to emitIdle, "keepAgent" to keepAgent, "hasCall" to (call != null), "hasAgent" to (callAgent != null), "recordsFailures" to (failures != null), ) // A dozen releases share one `try`, so the catch below cannot name the one that threw — and // the first throw skips every release after it. The breadcrumb is what turns "leave cleanup // failed" into a location; it is coarse on purpose, one name per group of related releases. var releasing = "telemetry" try { ticker.stop() detachDiagnostics() detachMediaStats() detachCapabilities() roster.detach() releasing = "audio" phoneMic.setEnabled(false) stopUplink() incomingPump.reset() incomingProbe.reset() applier.reset() scheduler.cancelPending() // Drop buffered voice before the dump: whatever is still in the chain belongs to a call that // is over, and the next call builds its own chain rather than inheriting this one. uplinkChain?.reset() uplinkChain = null externalPcmEnabled.set(false) pcmBridge?.finishDump() // Detach before stop so the teardown's own IDLE transition does not emit a // snapshot (or schedule a rebuild) for a call that is going away. releasing = "media" cancelMediaRestart() media.setStateListener(null) mediaSource = SourceState.IDLE currentSourceKind = SourceKind.WHEP media.stop() frameSender.detach() releasing = "none" } catch (error: Exception) { Log.w(TAG, "leave cleanup failed", error) // Whether this was recorded decides whether the host refuses the next call or never hears // about it, so the trace says which of the two happened. SoftApTrace.failure( "session_cleanup_failed", "releasing" to releasing, "reason" to "${error.javaClass.simpleName}: ${error.message ?: ""}", "recorded" to (failures != null), ) failures?.compareAndSet(null, error) } // Hang up and dispose must be independent: a failed hang-up must not skip // dispose, or the ACS agent leaks and the guest stays in the Teams roster. if (!keepAgent) { val hangUpStartedAt = SystemClock.elapsedRealtime() try { val pending = call?.hangUp() if (pending != null) { // Bounded on purpose: an unbounded `get()` on this Future is what parked Leave behind // the previous call on the single session executor, so Cancel never came back. pending.get(HANGUP_WAIT_MS, TimeUnit.MILLISECONDS) } SoftApTrace.stage( "session_hangup", "hadCall" to (call != null), "durationMs" to (SystemClock.elapsedRealtime() - hangUpStartedAt), ) } catch (timeout: TimeoutException) { Log.w(TAG, "leave hangUp timed out after ${HANGUP_WAIT_MS}ms") SoftApTrace.failure( "session_hangup_timeout", "durationMs" to (SystemClock.elapsedRealtime() - hangUpStartedAt), "timeoutMs" to HANGUP_WAIT_MS, "recorded" to (failures != null), ) failures?.compareAndSet(null, IllegalStateException("acs_hangup_timeout")) } catch (error: Exception) { Log.w(TAG, "leave hangUp failed", error) SoftApTrace.failure( "session_hangup_failed", "durationMs" to (SystemClock.elapsedRealtime() - hangUpStartedAt), "reason" to "${error.javaClass.simpleName}: ${error.message ?: ""}", "recorded" to (failures != null), ) failures?.compareAndSet(null, error) } val disposeStartedAt = SystemClock.elapsedRealtime() try { callAgent?.dispose() SoftApTrace.stage( "session_agent_disposed", "hadAgent" to (callAgent != null), "durationMs" to (SystemClock.elapsedRealtime() - disposeStartedAt), ) } catch (error: Exception) { Log.w(TAG, "leave dispose failed", error) // A leaked agent keeps the identity ACS refuses to share, so the next sign-in fails with // "CallAgent associated with this identity already exists" rather than here. SoftApTrace.failure( "session_agent_dispose_failed", "durationMs" to (SystemClock.elapsedRealtime() - disposeStartedAt), "reason" to "${error.javaClass.simpleName}: ${error.message ?: ""}", "recorded" to (failures != null), ) failures?.compareAndSet(null, error) } disposeAbandonedAgent(waitMs = 0) callClient = null call = null callAgent = null preparedToken = null preparedIdentityMode = null agentPrepared = false } media.stop() audioOut = null localOut = null audioIn = null videoOut = null videoEnabled = true outgoingReady.set(false) muted.set(false) hangUpForEveryone = CapabilityStatus() manageLobby = CapabilityStatus() audioSource = "glasses" lastSafety = AudioSafety.DEGRADED meetingUrl = null // Not `callOrigin`: a teardown trace still belongs to the call that is ending, and the next // join overwrites it before anything else is stamped. callId = "" lobbyEnteredAtMs = 0L connectedAtMs = 0L lastBweSampleAtMs = 0L wireEpisodes = WireEpisodeTracker() lastAdaptationAtMs = 0L lastLowWireAtMs = 0L // Clearing lastError is scoped to the clean idle reset. A failed join tears // down with emitIdle=false and relies on lastError staying set so emit("error") // still carries it and pushCallState keeps ignoring late disconnected callbacks. if (emitIdle) { lastError = null emit("idle") } SoftApTrace.stage( "session_cleanup_end", "durationMs" to (SystemClock.elapsedRealtime() - startedAt), "keptAgent" to keepAgent, "recordedFailure" to (failures?.get()?.let { "${it.javaClass.simpleName}: ${it.message ?: ""}" } ?: "none"), ) } private inner class SessionAudioController : AudioStreamController { override fun readActive(): ActiveStreamKind { val stream = call?.activeOutgoingAudioStream ?: return ActiveStreamKind.NONE if (stream.state != AudioStreamState.STARTED) return ActiveStreamKind.NONE return when (stream.type) { AudioStreamType.VIRTUAL_OUTGOING -> ActiveStreamKind.VIRTUAL AudioStreamType.LOCAL_OUTGOING -> ActiveStreamKind.LOCAL else -> ActiveStreamKind.NONE } } override fun isPhysicallyMuted(): Boolean? = call?.isOutgoingAudioMuted override fun setGlassesPcmEnabled(enabled: Boolean) { val routing = GlassesPcmRouting.decide(softap = currentSourceKind == SourceKind.SOFTAP, enabled = enabled) externalPcmEnabled.set(routing.externalPcm) media.setPcmDeliveryEnabled(routing.relayPcm) } override fun setPhonePcmEnabled(enabled: Boolean) { phoneMic.setEnabled(enabled) } override fun mutePhysical(): Result { val c = CallGuard.require(call).getOrElse { return Result.failure(it) } return runCatching { c.muteOutgoingAudio(context).get() } } override fun unmutePhysical(): Result { val c = CallGuard.require(call).getOrElse { return Result.failure(it) } return runCatching { c.unmuteOutgoingAudio(context).get() } } override fun stopActive(): Result { val c = CallGuard.require(call).getOrElse { return Result.failure(it) } val stream = c.activeOutgoingAudioStream ?: return Result.failure(IllegalStateException("no active stream")) return runCatching { c.stopAudio(context, stream).get() } } } companion object { private const val TAG = "ACS-SPIKE" const val GLASSES_REQUIRES_UNMUTED_TRANSPORT = true private const val ROSTER_COALESCE_MS = 150L private const val VIDEO_TOGGLE_TIMEOUT_MS = 10_000L private const val MEDIA_RESTART_BASE_MS = 1_000L private const val MEDIA_RESTART_MAX_MS = 10_000L /** * How many times to ask ACS for 1 Hz MEDIA_STATISTICS before giving up. * * With the backoff in [scheduleMediaStatsInterval] — 2 s, then 5 s, then 15 s — this spans * roughly eleven minutes, which is deliberately longer than a short call. The old budget was * six attempts over ten seconds and it never succeeded in any capture, because the condition * being waited on is a remote subscriber pulling the stream rather than a fixed startup delay. */ private const val MEDIA_STATS_INTERVAL_MAX_ATTEMPTS = 60 /** SoftAP join() blocks until the WHIP listener is bound and ACS join is queued. */ private const val SOFTAP_JOIN_WAIT_MS = 45_000L /** * How long [rebindSoftApIngest] waits for the parked listener to release its port after a * force-close. Not a tombstone: the old generation is closed immediately. */ private const val REBIND_INGEST_CLOSE_MS = 2_000L /** * How long End waits for ACS to accept the hang-up before reporting it unconfirmed. Local * teardown runs either way; this only bounds how long the wearer stares at a confirm sheet. */ /** * How long Leave waits for ACS `hangUp()` before disposing the agent anyway. * * This Future has no timeout of its own. On the single session executor an unbounded `get()` * is the hang that made Cancel sit on "Leaving the meeting" while the next join queued behind * the same stuck hang-up. */ private const val HANGUP_WAIT_MS = 8_000L private const val END_FOR_EVERYONE_WAIT_MS = 15_000L /** * How long to wait for ACS to hand back a call agent. * * Well inside [SOFTAP_JOIN_WAIT_MS] on purpose: this step needs the internet, and the hotspot * join just changed which network provides it. On device it stalled 30 s here and then blew the * whole join budget, so the wearer got one useless timeout instead of a nameable failure. */ private const val CALL_AGENT_WAIT_MS = 20_000L /** * How long [prepareAgent] waits, which is longer than [CALL_AGENT_WAIT_MS] and does not need to * fit any other budget: nothing is joined or held while it runs, so overrunning costs a slower * join rather than a hotspot the wearer then has to leave. * * Sized off a cold sign-in measured at ~35 s on a 544 ms-RTT AP. */ private const val PREPARE_AGENT_WAIT_MS = 60_000L /** How long to keep sweeping for an agent that arrives after its wait was abandoned. */ private const val LATE_AGENT_SWEEP_MS = 5_000L private const val LATE_AGENT_SWEEPS = 12 /** * How long a rejoin waits for the abandoned sign-in to finish so we can dispose it. Shorter * than [CALL_AGENT_WAIT_MS]: the leftover agent is usually already done, and blocking the * wearer again for a full sign-in would hide a stuck Future. */ private const val ABANDONED_AGENT_REJOIN_WAIT_MS = 8_000L fun sampleRateHz(rate: AudioStreamSampleRate?): Int? = when (rate) { AudioStreamSampleRate.HZ_16000 -> 16000 AudioStreamSampleRate.HZ_22050 -> 22050 AudioStreamSampleRate.HZ_24000 -> 24000 AudioStreamSampleRate.HZ_32000 -> 32000 AudioStreamSampleRate.HZ_44100 -> 44100 AudioStreamSampleRate.HZ_48000 -> 48000 null -> null } private fun describeEndReason(call: CommonCall?): Map { if (call == null) return mapOf("hasCall" to false) return try { val getter = call.javaClass.methods.firstOrNull { it.name == "getCallEndReason" || it.name == "getEndReason" } val reason = getter?.invoke(call) ?: return mapOf("hasCall" to true, "endReason" to "null") val methods = reason.javaClass.methods fun pick(vararg names: String): Any? = names.firstNotNullOfOrNull { name -> methods.firstOrNull { it.name == name && it.parameterCount == 0 }?.invoke(reason) } val message = pick("getMessage")?.toString()?.take(120) mapOf( "hasCall" to true, "code" to pick("getCode"), "subcode" to pick("getSubcode", "getSubCode"), "message" to message, ) } catch (error: Throwable) { mapOf("hasCall" to true, "endReasonError" to "${error.javaClass.simpleName}:${error.message ?: ""}") } } private fun formatAcsError(error: Throwable): String { val parts = linkedSetOf() var current: Throwable? = error var depth = 0 while (current != null && depth < 4) { val message = current.message?.trim().orEmpty() val piece = if (message.isNotEmpty()) "${current.javaClass.simpleName}: $message" else current.javaClass.simpleName parts.add(piece) current = current.cause depth += 1 } return parts.joinToString(" | ").ifBlank { "ACS join failed" } } } }