Skip to content

Commit ff277d5

Browse files
authored
Align agents-js with Python PR #4834 refactor and follow-up fixes (#1061)
1 parent cfe0362 commit ff277d5

10 files changed

Lines changed: 358 additions & 90 deletions

File tree

agents/src/inference/interruption/http_transport.ts

Lines changed: 14 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -120,7 +120,8 @@ export function createHttpTransport(
120120
}
121121

122122
const state = getState();
123-
if (!state.overlapSpeechStartedAt) return;
123+
const overlapSpeechStartedAt = state.overlapSpeechStartedAt;
124+
if (overlapSpeechStartedAt === undefined || !state.overlapSpeechStarted) return;
124125

125126
try {
126127
const resp = await predictHTTP(
@@ -135,16 +136,18 @@ export function createHttpTransport(
135136
);
136137

137138
const { createdAt, isBargein, probabilities, predictionDurationInS } = resp;
138-
const entry = new InterruptionCacheEntry({
139+
const entry = state.cache.setOrUpdate(
139140
createdAt,
140-
probabilities,
141-
isInterruption: isBargein,
142-
speechInput: chunk,
143-
totalDurationInS: (performance.now() - createdAt) / 1000,
144-
detectionDelayInS: (Date.now() - state.overlapSpeechStartedAt) / 1000,
145-
predictionDurationInS,
146-
});
147-
state.cache.set(createdAt, entry);
141+
() => new InterruptionCacheEntry({ createdAt }),
142+
{
143+
probabilities,
144+
isInterruption: isBargein,
145+
speechInput: chunk,
146+
totalDurationInS: (performance.now() - createdAt) / 1000,
147+
detectionDelayInS: (Date.now() - overlapSpeechStartedAt) / 1000,
148+
predictionDurationInS,
149+
},
150+
);
148151

149152
if (state.overlapSpeechStarted && entry.isInterruption) {
150153
if (updateUserSpeakingSpan) {
@@ -153,7 +156,7 @@ export function createHttpTransport(
153156
const event: InterruptionEvent = {
154157
type: InterruptionEventType.INTERRUPTION,
155158
timestamp: Date.now(),
156-
overlapSpeechStartedAt: state.overlapSpeechStartedAt,
159+
overlapSpeechStartedAt,
157160
isInterruption: entry.isInterruption,
158161
speechInput: entry.speechInput,
159162
probabilities: entry.probabilities,

agents/src/inference/interruption/interruption_cache_entry.ts

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -9,6 +9,7 @@ import { estimateProbability } from './utils.js';
99
*/
1010
export class InterruptionCacheEntry {
1111
createdAt: number;
12+
requestStartedAt?: number;
1213
totalDurationInS: number;
1314
predictionDurationInS: number;
1415
detectionDelayInS: number;
@@ -18,6 +19,7 @@ export class InterruptionCacheEntry {
1819

1920
constructor(params: {
2021
createdAt: number;
22+
requestStartedAt?: number;
2123
speechInput?: Int16Array;
2224
totalDurationInS?: number;
2325
predictionDurationInS?: number;
@@ -26,6 +28,7 @@ export class InterruptionCacheEntry {
2628
isInterruption?: boolean;
2729
}) {
2830
this.createdAt = params.createdAt;
31+
this.requestStartedAt = params.requestStartedAt;
2932
this.totalDurationInS = params.totalDurationInS ?? 0;
3033
this.predictionDurationInS = params.predictionDurationInS ?? 0;
3134
this.detectionDelayInS = params.detectionDelayInS ?? 0;

agents/src/inference/interruption/interruption_stream.ts

Lines changed: 24 additions & 21 deletions
Original file line numberDiff line numberDiff line change
@@ -48,14 +48,15 @@ export class InterruptionStreamSentinel {
4848
}
4949

5050
static overlapSpeechStarted(
51-
speechDurationInS: number,
51+
speechDuration: number,
52+
startedAt: number,
5253
userSpeakingSpan?: Span,
5354
): OverlapSpeechStarted {
54-
return { type: 'overlap-speech-started', speechDurationInS, userSpeakingSpan };
55+
return { type: 'overlap-speech-started', speechDuration, startedAt, userSpeakingSpan };
5556
}
5657

57-
static overlapSpeechEnded(): OverlapSpeechEnded {
58-
return { type: 'overlap-speech-ended' };
58+
static overlapSpeechEnded(endedAt: number): OverlapSpeechEnded {
59+
return { type: 'overlap-speech-ended', endedAt };
5960
}
6061

6162
static flush(): Flush {
@@ -226,18 +227,21 @@ export class InterruptionStreamBase {
226227
this.logger.debug('agent speech started');
227228
agentSpeechStarted = true;
228229
overlapSpeechStarted = false;
230+
this.overlapSpeechStartedAt = undefined;
229231
accumulatedSamples = 0;
230232
startIdx = 0;
231233
cache.clear();
232234
} else if (chunk.type === 'agent-speech-ended') {
233235
this.logger.debug('agent speech ended');
234236
agentSpeechStarted = false;
235237
overlapSpeechStarted = false;
238+
this.overlapSpeechStartedAt = undefined;
236239
accumulatedSamples = 0;
237240
overlapCount = 0;
238241
startIdx = 0;
239242
cache.clear();
240243
} else if (chunk.type === 'overlap-speech-started' && agentSpeechStarted) {
244+
this.overlapSpeechStartedAt = chunk.startedAt;
241245
this.userSpeakingSpan = chunk.userSpeakingSpan;
242246
this.logger.debug('overlap speech started, starting interruption inference');
243247
overlapSpeechStarted = true;
@@ -247,13 +251,13 @@ export class InterruptionStreamBase {
247251
// leading silence) when the first overlap speech started.
248252
// Otherwise, keep the existing data.
249253
if (overlapCount <= 1) {
250-
const shiftSize = Math.min(
251-
startIdx,
252-
Math.round(chunk.speechDurationInS * this.options.sampleRate) +
253-
Math.round(this.options.audioPrefixDurationInS * this.options.sampleRate),
254-
);
255-
inferenceS16Data.copyWithin(0, startIdx - shiftSize, startIdx);
256-
startIdx = shiftSize;
254+
const keepSize =
255+
// Convert speechDuration (ms) → samples; audioPrefixDurationInS (s) → samples
256+
Math.round((chunk.speechDuration / 1000) * this.options.sampleRate) +
257+
Math.round(this.options.audioPrefixDurationInS * this.options.sampleRate);
258+
const shiftCount = Math.max(0, startIdx - keepSize);
259+
inferenceS16Data.copyWithin(0, shiftCount, startIdx);
260+
startIdx -= shiftCount;
257261
}
258262
cache.clear();
259263
} else if (chunk.type === 'overlap-speech-ended') {
@@ -269,22 +273,24 @@ export class InterruptionStreamBase {
269273
this.logger.debug('no request made for overlap speech');
270274
latestEntry = InterruptionCacheEntry.default();
271275
}
276+
const latestEntryValue = latestEntry ?? InterruptionCacheEntry.default();
272277
const event: InterruptionEvent = {
273278
type: InterruptionEventType.OVERLAP_SPEECH_ENDED,
274-
timestamp: Date.now(),
279+
timestamp: chunk.endedAt,
275280
isInterruption: false,
276281
overlapSpeechStartedAt: this.overlapSpeechStartedAt,
277-
speechInput: latestEntry.speechInput,
278-
probabilities: latestEntry.probabilities,
279-
totalDurationInS: latestEntry.totalDurationInS,
280-
detectionDelayInS: latestEntry.detectionDelayInS,
281-
predictionDurationInS: latestEntry.predictionDurationInS,
282-
probability: latestEntry.probability,
282+
speechInput: latestEntryValue.speechInput,
283+
probabilities: latestEntryValue.probabilities,
284+
totalDurationInS: latestEntryValue.totalDurationInS,
285+
detectionDelayInS: latestEntryValue.detectionDelayInS,
286+
predictionDurationInS: latestEntryValue.predictionDurationInS,
287+
probability: latestEntryValue.probability,
283288
};
284289
controller.enqueue(event);
285290
overlapSpeechStarted = false;
286291
accumulatedSamples = 0;
287292
}
293+
this.overlapSpeechStartedAt = undefined;
288294
} else if (chunk.type === 'flush') {
289295
// no-op
290296
}
@@ -349,9 +355,6 @@ export class InterruptionStreamBase {
349355
async pushFrame(frame: InterruptionSentinel | AudioFrame): Promise<void> {
350356
this.ensureStreamsNotEnded();
351357
if (!(frame instanceof AudioFrame)) {
352-
if (frame.type === 'overlap-speech-started') {
353-
this.overlapSpeechStartedAt = Date.now() - frame.speechDurationInS * 1000;
354-
}
355358
return this.inputStream.write(frame);
356359
} else if (this.options.sampleRate !== frame.sampleRate) {
357360
const resampler = this.getResamplerFor(frame.sampleRate);

agents/src/inference/interruption/types.ts

Lines changed: 6 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -66,12 +66,17 @@ export interface AgentSpeechEnded {
6666

6767
export interface OverlapSpeechStarted {
6868
type: 'overlap-speech-started';
69-
speechDurationInS: number;
69+
/** Duration of the speech segment in milliseconds (matches VADEvent.speechDuration units). */
70+
speechDuration: number;
71+
/** Absolute timestamp (ms) when overlap speech started, computed at call-site. */
72+
startedAt: number;
7073
userSpeakingSpan?: Span;
7174
}
7275

7376
export interface OverlapSpeechEnded {
7477
type: 'overlap-speech-ended';
78+
/** Absolute timestamp (ms) when overlap speech ended, used as the non-interruption event timestamp. */
79+
endedAt: number;
7580
}
7681

7782
export interface Flush {
Lines changed: 132 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,132 @@
1+
// SPDX-FileCopyrightText: 2026 LiveKit, Inc.
2+
//
3+
// SPDX-License-Identifier: Apache-2.0
4+
import { describe, expect, it, vi } from 'vitest';
5+
import { BoundedCache } from './utils.js';
6+
7+
class Entry {
8+
createdAt: number;
9+
totalDurationInS: number | undefined = undefined;
10+
predictionDurationInS: number | undefined = undefined;
11+
note: string | undefined = undefined;
12+
13+
constructor(createdAt: number, note?: string) {
14+
this.createdAt = createdAt;
15+
this.note = note;
16+
}
17+
}
18+
19+
describe('BoundedCache', () => {
20+
it('evicts oldest entry when maxLen is exceeded', () => {
21+
const cache = new BoundedCache<number, Entry>(2);
22+
cache.set(1, new Entry(1));
23+
cache.set(2, new Entry(2));
24+
cache.set(3, new Entry(3));
25+
26+
expect(cache.size).toBe(2);
27+
expect([...cache.keys()]).toEqual([2, 3]);
28+
expect(cache.get(1)).toBeUndefined();
29+
expect(cache.get(2)!.createdAt).toBe(2);
30+
expect(cache.get(3)!.createdAt).toBe(3);
31+
});
32+
33+
it('setOrUpdate creates a value via factory when key is missing', () => {
34+
const cache = new BoundedCache<number, Entry>(10);
35+
const factory = vi.fn(() => new Entry(100));
36+
37+
const value = cache.setOrUpdate(1, factory, { predictionDurationInS: 0.42 });
38+
39+
expect(factory).toHaveBeenCalledTimes(1);
40+
expect(value.createdAt).toBe(100);
41+
expect(value.predictionDurationInS).toBe(0.42);
42+
expect(cache.get(1)?.predictionDurationInS).toBe(0.42);
43+
});
44+
45+
it('setOrUpdate updates existing value and does not call factory', () => {
46+
const cache = new BoundedCache<number, Entry>(10);
47+
cache.set(1, new Entry(1, 'before'));
48+
const factory = vi.fn(() => new Entry(999));
49+
50+
const value = cache.setOrUpdate(1, factory, { note: 'after', totalDurationInS: 1.5 });
51+
52+
expect(factory).not.toHaveBeenCalled();
53+
expect(value.createdAt).toBe(1);
54+
expect(value.note).toBe('after');
55+
expect(value.totalDurationInS).toBe(1.5);
56+
});
57+
58+
it('updateValue returns undefined for missing key', () => {
59+
const cache = new BoundedCache<number, Entry>(10);
60+
const result = cache.updateValue(404, { note: 'missing' });
61+
62+
expect(result).toBeUndefined();
63+
});
64+
65+
it('updateValue ignores undefined fields', () => {
66+
const cache = new BoundedCache<number, Entry>(10);
67+
cache.set(1, new Entry(1, 'keep'));
68+
69+
const result = cache.updateValue(1, {
70+
note: undefined,
71+
predictionDurationInS: 0.1,
72+
});
73+
74+
expect(result?.createdAt).toBe(1);
75+
expect(result?.note).toBe('keep');
76+
expect(result?.predictionDurationInS).toBe(0.1);
77+
});
78+
79+
it('pop without predicate removes the oldest entry (python parity)', () => {
80+
const cache = new BoundedCache<number, Entry>(10);
81+
cache.set(1, new Entry(1));
82+
cache.set(2, new Entry(2));
83+
cache.set(3, new Entry(3));
84+
85+
const popped = cache.pop();
86+
87+
expect(popped?.createdAt).toBe(1);
88+
expect([...cache.keys()]).toEqual([2, 3]);
89+
});
90+
91+
it('pop with predicate removes the most recent matching entry', () => {
92+
const cache = new BoundedCache<number, Entry>(10);
93+
const e1 = new Entry(1);
94+
e1.totalDurationInS = 0;
95+
const e2 = new Entry(2);
96+
e2.totalDurationInS = 1;
97+
const e3 = new Entry(3);
98+
e3.totalDurationInS = 2;
99+
cache.set(1, e1);
100+
cache.set(2, e2);
101+
cache.set(3, e3);
102+
103+
const popped = cache.pop((entry) => (entry.totalDurationInS ?? 0) > 0);
104+
105+
expect(popped?.createdAt).toBe(3);
106+
expect(popped?.totalDurationInS).toBe(2);
107+
expect([...cache.keys()]).toEqual([1, 2]);
108+
});
109+
110+
it('pop with predicate returns undefined when no match exists', () => {
111+
const cache = new BoundedCache<number, Entry>(10);
112+
const e1 = new Entry(1);
113+
e1.totalDurationInS = 0;
114+
cache.set(1, e1);
115+
116+
const popped = cache.pop((entry) => (entry.totalDurationInS ?? 0) > 10);
117+
118+
expect(popped).toBeUndefined();
119+
expect(cache.size).toBe(1);
120+
});
121+
122+
it('clear removes all entries', () => {
123+
const cache = new BoundedCache<number, Entry>(10);
124+
cache.set(1, new Entry(1));
125+
cache.set(2, new Entry(2));
126+
127+
cache.clear();
128+
129+
expect(cache.size).toBe(0);
130+
expect([...cache.keys()]).toEqual([]);
131+
});
132+
});

0 commit comments

Comments
 (0)