fix(voice): preserve request ownership across callbacks

keep typed and tool-follow-up responses attached to their original requests
reject duplicate events before they affect newer work
preserve newer input and Radio handoffs during old callback cleanup
integrate the tested main snapshot without dropping existing history
This commit is contained in:
manjunath committed 2026-10-16 22:29:00 +05:30
commit 8abc99aa0a
26 files changed
+4010 -132

No files matched your search

+27
View File
@@ -1,5 +1,32 @@
# Changelog
- Add Bogotá (Tembici, 250 stations) and Santiago de Chile (Bike Itaú, 200
stations) to the GBFS bikeshare registry, the layer's first stations in
Colombia and Chile. Both are PBSC feeds like the Brazilian Bike Itaú cities,
so no proxy change is needed. Both feeds also list depot stations on other
continents (São Paulo, Madrid) as in service; stations outside a city's load
radius are now dropped instead of drawn as live docks.
- Preserve newer typed input submitted during rejected-response cleanup, and
reject repeated OpenAI response creation IDs before they can consume a newer
speech turn or alter response state. Release exact ID history on Stop/reset.
- Correlate OpenAI typed and tool-follow-up responses with client-authored
request metadata instead of automatic speech-response arrival order. Keep
deferred responses tied to their original input through send callbacks.
- Ignore duplicate or previously admitted OpenAI speech-start events before
retiring input, cancelling tools or dropping queued typed responses. Retain
input IDs until session Stop/reset, and prevent observer-triggered Stop or
newer input from being overwritten by the interrupted admission callback.
An onset superseded inside a notification retains only a retired identity
for its later commit, keeping its response separate from newer typed input.
- Preserve typed command ownership when an unobserved Gemini automatic activity
receives its interruption boundary; a separate new speech onset still revokes
that typed request during the wait. A newer OpenAI speech turn drops an unsent
typed response instead of binding that response to the speech transcript.
- Add Barcelona's bike share to the GBFS registry: Bicing (543 stations, the
city's PBSC system) and AMBici (237 e-bike stations across the metropolitan
area, run by nextbike). Both are keyless GBFS 2.3 feeds listed in
+1 -1
View File
@@ -67,7 +67,7 @@ How to read this:
| **City of Toronto Open Data** (`opendata.toronto.ca`) | CCTV camera catalog + frames, Toronto city streets | [Open Government Licence – Toronto](https://www.toronto.ca/city-government/data-research-maps/open-data/open-data-licence/) — attribution REQUIRED | "Contains information licensed under the Open Government Licence – Toronto" |
| **GBFS (Lyft / BCycle / Social Bicycles / PBSC)** | Bikeshare availability | Per-feed (attribution-only) | Credit the operator (e.g. Austin BCycle) + its `license_url` |
| **Bike Share Toronto** (`toronto.publicbikesystem.net`) | Bikeshare station availability, Toronto | Operator GBFS feed. No licence is published: the City's open-data package declares none and the feed has no `license_url`. Attribution is given to the operator. | "Bike Share Toronto (Toronto Parking Authority)" |
| **Bike Itaú (Tembici)** (`*.publicbikesystem.net`, seven Brazilian cities) | Bikeshare station availability: São Paulo, Rio de Janeiro, Salvador, Recife, Porto Alegre, Brasília and Curitiba | Operator GBFS feeds. No licence is published: the feeds' `system_information.json` carries no `license_url`, operator or attribution fields. Attribution is given to the operator. | "Bike Itaú (Tembici)" |
| **Bike Itaú / Tembici** (`*.publicbikesystem.net`, seven Brazilian cities, Santiago and Bogotá) | Bikeshare station availability: São Paulo, Rio de Janeiro, Salvador, Recife, Porto Alegre, Brasília, Curitiba, Santiago de Chile and Bogotá | Operator GBFS feeds. No licence is published: the feeds' `system_information.json` carries no `license_url`, operator or attribution fields. Attribution is given to the operator. | "Bike Itaú (Tembici)"; "Tembici Bogotá" |
| **Bicing** (`barcelona.publicbikesystem.net`) | Bikeshare station availability, Barcelona | Operator GBFS feed (PBSC), keyless. The feed publishes no licence; the City's own open-data copy of the same station data is CC BY 4.0 but needs a token. Attribution is given to the operator. | "Bicing, Barcelona City Council (B:SM)" |
| **AMBici** (nextbike, `gbfs.nextbike.net`) | Bikeshare station availability, Barcelona metropolitan area (e-bikes) | Operator GBFS feed, keyless. `system_information.license_id` is **CC0-1.0**; no attribution required, given anyway. | "AMBici, Àrea Metropolitana de Barcelona, operated by nextbike" |
| **Oslo / Bergen / Trondheim Bysykkel** (Urban Sharing, `gbfs.urbansharing.com`) | Bikeshare availability, Oslo, Bergen and Trondheim city bikes. Keyless GBFS; the proxy sends the `Client-Identifier` header the operators ask for | [NLOD 2.0](https://data.norge.no/nlod/en/2.0) — attribution REQUIRED ([Oslo](https://oslobysykkel.no/en/open-data/realtime), [Bergen](https://bergenbysykkel.no/en/apne-data/sanntid), [Trondheim](https://trondheimbysykkel.no/en/open-data/realtime)) | "Contains data under the Norwegian licence for Open Government data (NLOD) made available by Oslo Bysykkel, Bergen Bysykkel and Trondheim Bysykkel" |
+42
View File
@@ -7,6 +7,29 @@ deferred Radio playback; completed layer changes or station selection remain
reported truthfully. This limit refusal does not infer a provider request or
change providers.
OpenAI admits each nonempty provider speech-input ID once per owned connection
lifetime, before input, tool, narration or response-queue effects. Stop/reset
clears that exact admission history; storage grows with admitted inputs and ID
bytes. A newer local input or Stop from an observer prevents the older onset
from overwriting it. An admitted onset superseded before becoming current is
remembered only as retired for its later real commit: it grants no authority
and reserves no response early. Existing input-record and response-queue limits
are unchanged.
OpenAI explicitly requested typed and tool-follow-up responses carry a client
request ID in response metadata. Their returned IDs select only that original
request; untagged automatic responses use speech-commit slots and cannot borrow
typed authority. Unknown or malformed client IDs fail stale. Deferred requests
retain their input identity, and Stop or newer input invalidates old captures.
A tool follow-up retains the requesting response's record rather than borrowing
another current input. Gemini keeps its separate existing response binding.
OpenAI response-creation IDs are also admitted once per owned lifetime, before
binding, narration, metrics or response bookkeeping. Other distinct events for
the response continue normally. This exact history grows with response count
and ID bytes and is cleared on Stop/reset. Rejection cleanup completes before
callbacks that may submit newer input, preserving that input without retrying
the rejected request.
Voice retains authentic user-turn identity when OpenAI input transcription is
explicitly disabled or fails. Gemini native tool calls use current socket,
session, local-turn and call ownership; unordered input captions are advisory
@@ -5734,6 +5757,25 @@ when their footprints touch or contain one another, never across a seam toleranc
### Gemini Live provider integration
OpenAI admits each nonempty provider speech item ID once per connection lifetime,
so duplicate or previously admitted starts cannot retire current input, abort its
tools, restart narration or drop a queued typed request. The exact ID history is
retained until Stop/reset rather than evicting older IDs; memory grows with the
number and total size of admitted IDs in that lifetime. New unseen input still
supersedes earlier work. Observer-triggered Stop, replacement or newer typed
input cannot be overwritten when the earlier onset callback resumes.
Typed input keeps its own request identity across an interruption acknowledgment
for a locally known, unobserved Gemini automatic activity. This single-use
predecessor record does not treat captions as authority or make VAD alone wait
for a generated reply. It expires on successor output, completion or newer input;
later unattributed interruption markers retain the existing interruption policy.
A genuine new automatic speech onset also revokes typed input when it arrives
separately during that predecessor wait, without adding another retirement wait.
On OpenAI, a newer speech onset discards a typed response still waiting to be sent;
a deferred typed response with no newer input still answers once. The discarded
request may go unanswered if the superseding speech is noise or gets no response.
Standalone voice supports OpenAI and Gemini; embedded hosts default to OpenAI and opt into Gemini through `voice.supportedProviders`. The provider dropdown uses the existing microphone control and result card. Manual selection reconnects an active session in its prior input mode; push-to-talk stays muted until a fresh Space hold. Voice that was off stays off.
Gemini uses a separate server-side `GEMINI_API_KEY` and one-session ephemeral credentials from `/api/gemini/token`. Device-rate microphone capture is resampled to 16 kHz; playback uses a separate 24 kHz context. Microphone permission precedes token minting. Bounded control queues, retired replies and pending tool actions are invalidated on interruption, Stop, replacement and teardown; input capture identities follow the physical microphone gate and end on Stop, replacement and teardown rather than on every interruption. `goAway` stops new turns and drains the current reply until completion or its supplied deadline, then requires an explicit restart.
+1 -1
View File
@@ -304,7 +304,7 @@ export const DATA_CREDITS = [
key: 'gbfs',
html:
'Bikeshare availability: GBFS operator feeds ' +
'(e.g. Austin BCycle; Bike Share Toronto, operated by the Toronto Parking Authority; Bike Itaú, operated by Tembici; ' +
'(e.g. Austin BCycle; Bike Share Toronto, operated by the Toronto Parking Authority; Bike Itaú and Tembici Bogotá, operated by Tembici; ' +
'Bicing, Barcelona City Council (B:SM); AMBici, Àrea Metropolitana de Barcelona, operated by nextbike)',
},
{
+4 -1
View File
@@ -97,7 +97,10 @@ export function createIngestion({
const payload = await fetchGbfsJson(city.stationInformationUrl, {
signal: controller.signal,
});
const stationMap = parts.model.parseStationInformation(payload);
const stationMap = parts.model.stationsWithinCity(
city,
parts.model.parseStationInformation(payload),
);
if (stationMap.size === 0) {
throw new Error(`No station information for ${city.city}`);
}
@@ -0,0 +1,174 @@
import assert from 'node:assert/strict';
import test from 'node:test';
import { CITY_BY_ID } from './registry.js';
import { createIngestion } from './ingestion.js';
import { createModel } from './model.js';
import { createViewport } from './viewport.js';
import { isAllowedGbfsHost, isAllowedGbfsPath } from '../../data/gbfsSource.js';
// Each system's PBSC host, the station farthest from its registered centre,
// and the depots its live feed lists on other continents (checked 2026-10-16).
const CITIES = {
'bogota-tembici': {
host: 'bogota.publicbikesystem.net',
provider: 'Tembici Bogotá',
stations: [
{ name: '067 - KR 7B Bis con CL 152A', lat: 4.727729, lon: -74.026271 },
],
depots: [
{
station_id: '316',
name: 'Calle 13 - 316',
lat: -23.4008649,
lon: -46.3972181,
},
],
},
'santiago-bike-itau': {
host: 'santiago.publicbikesystem.net',
provider: 'Bike Itaú Santiago',
stations: [
{ name: 'LC077 - Mall Sport', lat: -33.3707813, lon: -70.5073122 },
],
depots: [
{
station_id: '459',
name: 'Galpon - 459',
lat: 40.4167747,
lon: -3.7037899,
},
{
station_id: '440',
name: 'Galpon - 440',
lat: -23.5690975,
lon: -46.697366,
},
],
},
};
// The production range check, with no viewer or fetches behind it.
const parts = {
model: createModel({ state: {}, services: {}, parts: {}, source: null }),
};
const { computeInRangeCities } = createViewport({
state: {},
services: { render: { governorRequestRender() {} } },
parts,
source: null,
});
test('Bogotá and Santiago are registered and readable through the proxy', () => {
for (const [id, expected] of Object.entries(CITIES)) {
const city = CITY_BY_ID.get(id);
assert.ok(city, `${id} is registered`);
assert.equal(city.provider, expected.provider);
assert.deepEqual(city.hosts, [expected.host]);
for (const feed of [city.stationInformationUrl, city.stationStatusUrl]) {
const url = new URL(feed);
assert.ok(isAllowedGbfsHost(url.hostname), `${id} host ${url.hostname}`);
assert.ok(isAllowedGbfsPath(url.pathname), `${id} path ${url.pathname}`);
}
}
});
test('a camera over each outermost station activates its system', () => {
for (const [id, expected] of Object.entries(CITIES)) {
for (const station of expected.stations) {
assert.ok(
computeInRangeCities({ lat: station.lat, lon: station.lon }).has(id),
`${station.name} activates ${id}`,
);
}
}
});
test('each load radius reaches its outermost stations', () => {
// The range check floors every radius at 100 km; the station filter and
// get_bike_share use the radius as registered.
for (const [id, expected] of Object.entries(CITIES)) {
const city = CITY_BY_ID.get(id);
for (const station of expected.stations) {
const distance = parts.model.haversineKm(
city.centerLat,
city.centerLon,
station.lat,
station.lon,
);
assert.ok(
distance <= city.loadRadiusKm,
`${station.name} is ${distance.toFixed(1)} km from the ${id} centre`,
);
}
}
});
test('stations a feed lists on another continent are not loaded', async () => {
for (const [id, expected] of Object.entries(CITIES)) {
const city = CITY_BY_ID.get(id);
const outermost = expected.stations[0];
const payload = {
data: {
stations: [
{
station_id: 'local',
name: outermost.name,
lat: outermost.lat,
lon: outermost.lon,
},
...expected.depots,
],
},
};
const state = {
_stationInfoCache: new Map(),
_inFlightInfo: new Map(),
_inFlightStatus: new Map(),
_loadingOps: 0,
_loading: false,
};
const ingestion = createIngestion({
state,
services: {},
parts,
source: {
async getStations(url) {
assert.equal(url, city.stationInformationUrl);
return payload;
},
},
});
const stations = await ingestion.loadCityStationInfo(id, 1);
assert.deepEqual(
[...stations.keys()],
['local'],
`${id} keeps only local stations`,
);
assert.equal(state._stationInfoCache.get(id), stations);
}
});
test('a feed with only far-away stations reports no station information', async () => {
const city = CITY_BY_ID.get('santiago-bike-itau');
const ingestion = createIngestion({
state: {
_stationInfoCache: new Map(),
_inFlightInfo: new Map(),
_inFlightStatus: new Map(),
_loadingOps: 0,
_loading: false,
},
services: {},
parts,
source: {
async getStations() {
return { data: { stations: CITIES['santiago-bike-itau'].depots } };
},
},
});
await assert.rejects(
ingestion.loadCityStationInfo('santiago-bike-itau', 1),
new RegExp(`No station information for ${city.city}`),
);
});
+25
View File
@@ -39,11 +39,36 @@ export function createModel({ state: layerState, services, parts, source }) {
return 6371 * 2 * Math.atan2(Math.sqrt(a), Math.sqrt(1 - a));
}
/**
* Drop stations outside a city's load radius. Some operators list depots on
* other continents as in-service stations in a city's feed (Bogotá's and
* Santiago's PBSC feeds carry warehouses in São Paulo and Madrid), which
* would otherwise draw as live docks thousands of kilometres away.
* @param {{ centerLat: number, centerLon: number, loadRadiusKm: number }} city
* @param {Map<string, Object>} stationMap - Parsed station info map.
* @returns {Map<string, Object>} Stations within the city's load radius.
*/
function stationsWithinCity(city, stationMap) {
const kept = new Map();
for (const [stationId, station] of stationMap) {
const distance = haversineKm(
city.centerLat,
city.centerLon,
station.lat,
station.lon,
);
if (distance <= city.loadRadiusKm) kept.set(stationId, station);
}
return kept;
}
return {
normalizeGbfsBool,
toNonNegativeInteger,
stationKey,
haversineKm,
stationsWithinCity,
extractStationsArray,
parseStationInformation,
parseStationStatus,
+24
View File
@@ -330,6 +330,30 @@ const RAW_GBFS_CITY_REGISTRY = [
'https://curitiba.publicbikesystem.net/customer/gbfs/v2/en/station_status.json',
provider: 'Bike Itaú',
},
{
id: 'bogota-tembici',
city: 'Bogotá, Colombia',
centerLat: 4.6533,
centerLon: -74.0836,
loadRadiusKm: 90,
stationInformationUrl:
'https://bogota.publicbikesystem.net/customer/gbfs/v2/en/station_information.json',
stationStatusUrl:
'https://bogota.publicbikesystem.net/customer/gbfs/v2/en/station_status.json',
provider: 'Tembici Bogotá',
},
{
id: 'santiago-bike-itau',
city: 'Santiago, Chile',
centerLat: -33.4372,
centerLon: -70.6506,
loadRadiusKm: 90,
stationInformationUrl:
'https://santiago.publicbikesystem.net/customer/gbfs/v2/en/station_information.json',
stationStatusUrl:
'https://santiago.publicbikesystem.net/customer/gbfs/v2/en/station_status.json',
provider: 'Bike Itaú Santiago',
},
{
id: 'toronto-bike-share',
city: 'Toronto, ON',
+152 -30
View File
@@ -153,6 +153,7 @@ export class GeminiSessionController {
this.model = null;
this.epoch = 0;
this.turn = 0;
this.inputGeneration = 0;
this.turnAbort = new AbortController();
this.disposed = false;
this.socket = null;
@@ -460,12 +461,14 @@ export class GeminiSessionController {
writeControl(entry) {
this.socket.send(entry.body);
if (!this.controlCurrent(entry)) return;
const input = entry.message.realtimeInput;
// Only markers actually written to the socket own its physical activity.
if (input?.activityStart) {
this.forwardedActivityOpen = true;
this.forwardedCaptureId = entry.captureId;
this.beginSpeechIntent(entry.captureId);
if (!this.controlCurrent(entry)) return;
}
if (input?.activityEnd) {
this.forwardedActivityOpen = false;
@@ -598,10 +601,24 @@ export class GeminiSessionController {
return;
if (!pushToTalk || activityCaptureId === metadata.captureId) return;
this.captureOwner.admitted = true;
this.interrupt('push-to-talk', {
keepPlayback: !this.pttPlaybackSuppressed,
});
this.beginPointerTurn('speech_start', this.captureOwner.snapshot);
if (
!this.interrupt('push-to-talk', {
keepPlayback: !this.pttPlaybackSuppressed,
})
)
return;
if (!this.owns(epoch) || !this.acceptsCapture(metadata)) return;
const activityGeneration = this.inputGeneration;
const activityCurrent = () =>
this.owns(epoch) &&
this.inputGeneration === activityGeneration &&
this.acceptsCapture(metadata);
this.beginPointerTurn(
'speech_start',
this.captureOwner.snapshot,
activityCurrent,
);
if (!activityCurrent()) return;
const context = this.pendingPointerContext;
this.captureOwner.context = context;
if (!this.sendPointerContext(context, metadata)) {
@@ -613,10 +630,12 @@ export class GeminiSessionController {
);
return;
}
if (!activityCurrent()) return;
const accepted = this.sendUserInput(
{ realtimeInput: { activityStart: {} } },
metadata,
);
if (!activityCurrent()) return;
// Quiet context and an empty hold have no response to retire later.
if (accepted) activityCaptureId = metadata.captureId;
else if (this.owns(epoch))
@@ -864,23 +883,21 @@ export class GeminiSessionController {
}
abortTools() {
this.narration?.cancel();
// A completed tool can still own progressive work (for example outlines).
// Retain its turn signal until an actual input/session boundary revokes it.
this.turnAbort.abort();
const resources = this.turnAbort;
const tools = [...this.tools.values()];
this.turnAbort = new AbortController();
this.turn++;
this.typedIntent = null;
this.providerQuarantine = null;
this.radioStrongerCommit = null;
this.pruneControls();
for (const entry of this.tools.values()) {
entry.abort();
}
this.tools.clear();
this.radio?.clearTools();
this.pendingTools = 0;
this.toolQueue = Promise.resolve();
this.narration?.cancel();
resources.abort();
for (const entry of tools) entry.abort();
}
noteUserActivity() {
@@ -912,8 +929,7 @@ export class GeminiSessionController {
!this.radio.mayVoiceClaimSpeaker()
)
return false;
this.interrupt('push-to-talk');
return true;
return this.interrupt('push-to-talk');
}
interrupt(
@@ -924,7 +940,14 @@ export class GeminiSessionController {
serverInterrupted = false,
} = {},
) {
if (this.goAway && !serverInterrupted) return;
if (this.goAway && !serverInterrupted) return false;
const epoch = this.epoch;
const generation = ++this.inputGeneration;
const channel = this.channel;
const owned = () =>
this.owns(epoch) &&
this.inputGeneration === generation &&
this.channel === channel;
this.pttPlaybackSuppressed = false;
if (['push-to-talk', 'user-text'].includes(reason))
this.manualInputKind = reason;
@@ -946,11 +969,14 @@ export class GeminiSessionController {
}
this.clearDeferredInput();
this.abortTools();
if (!owned()) return false;
this.userTurns.invalidate();
if (!owned()) return false;
this.intentItemId = null;
this.intentResponseId = null;
this.intentCaptureId = null;
this.emit({ type: 'user-turn-start' });
if (!owned()) return false;
if (deferInput) {
// Only one successor may reach Live before the previous boundary. A
// newer local turn replaces unsent input and waits to interrupt that
@@ -984,7 +1010,9 @@ export class GeminiSessionController {
owner.timer.unref?.();
} else this.boundRetiredReply();
this.noteUserActivity();
if (!owned()) return false;
if (!keepPlayback) this.audio?.clearPlayback();
if (!owned()) return false;
this.modelComplete = false;
this.pendingCompletionStatus = null;
this.postToolOutput = false;
@@ -995,13 +1023,16 @@ export class GeminiSessionController {
reason,
...(preserveAdvisoryCaption ? { preserveAdvisoryCaption: true } : {}),
});
if (!owned()) return false;
this.input.setVoiceSpeaker(this.goAway ? 'idle' : 'user');
if (!owned()) return false;
this.setStatus(
'listening',
this.goAway
? 'Gemini session expiring; finishing the current reply'
: 'Ask or command',
);
return owned();
}
handleMessage(message, epoch = this.epoch) {
@@ -1050,11 +1081,34 @@ export class GeminiSessionController {
// A server acknowledgment of a local interruption belongs to the retired
// reply; it must not revoke the successor's pointer, timing or turn token.
if (content?.interrupted) {
if (!this.retiredReplyPending)
this.interrupt('user-speech', {
preserveAdvisoryCaption: true,
serverInterrupted: true,
});
const newAutomaticActivity =
!this.input.pushToTalkMode &&
!this.goAway &&
message.voiceActivity?.type === 'ACTIVITY_START' &&
!this.userActivityOpen;
const predecessor = this.typedIntent?.automaticPredecessor;
if (
!this.retiredReplyPending &&
!newAutomaticActivity &&
predecessor?.epoch === epoch &&
this.readTypedTurnInput()
) {
// Automatic VAD may not have produced output before text superseded
// it. This specific local predecessor owns one interruption boundary;
// the acknowledgment is not another user input. Do not infer this
// from output silence alone, or wait for a boundary until it arrives.
this.retiredAutomaticActivityOwner = predecessor;
this.typedIntent.retiredAutomaticPredecessor = predecessor;
this.typedIntent.automaticPredecessor = null;
} else if (!this.retiredReplyPending) {
if (
!this.interrupt('user-speech', {
preserveAdvisoryCaption: true,
serverInterrupted: true,
})
)
return;
}
this.retiredReplyPending = true;
this.boundRetiredReply();
}
@@ -1066,7 +1120,19 @@ export class GeminiSessionController {
!this.userActivityOpen
) {
this.userActivityOpen = true;
if (!this.retiredReplyPending) this.interrupt();
const typedPredecessor = this.typedIntent?.retiredAutomaticPredecessor;
if (
this.retiredReplyPending &&
typedPredecessor &&
typedPredecessor === this.retiredAutomaticActivityOwner &&
this.readTypedTurnInput()
) {
// This precise wait still belongs to the automatic predecessor, not
// the submitted text. New speech revokes that text without inventing
// a second retirement/deferred input or renewing the existing bound.
this.replyPending = false;
if (!this.interrupt()) return;
} else if (!this.retiredReplyPending && !this.interrupt()) return;
// Local activity ownership is only for expiry liveness, not tools.
// Assign after retiring observed output so its boundary cannot clear
// a newer detected activity that has not received a reply yet.
@@ -1075,12 +1141,20 @@ export class GeminiSessionController {
sequence: ++this.automaticActivitySequence,
};
this.automaticTurnOpen = true;
const activityGeneration = this.inputGeneration;
const activityCurrent = () =>
this.owns(epoch) && this.inputGeneration === activityGeneration;
this.beginSpeechIntent(null);
this.beginPointerTurn('speech_start');
if (!activityCurrent()) return;
this.beginPointerTurn('speech_start', undefined, activityCurrent);
if (!activityCurrent()) return;
const context = this.pendingPointerContext;
if (
context &&
this.send({ realtimeInput: { text: JSON.stringify(context.item) } })
this.send({
realtimeInput: { text: JSON.stringify(context.item) },
}) &&
activityCurrent()
)
this.consumePointerContext(context);
} else if (
@@ -1129,6 +1203,11 @@ export class GeminiSessionController {
if (content?.turnComplete) {
this.clearRetiredBoundaryWait();
this.retiredReplyPending = false;
if (
this.typedIntent?.retiredAutomaticPredecessor ===
this.retiredAutomaticActivityOwner
)
this.typedIntent.retiredAutomaticPredecessor = null;
// A true replyPending now belongs to a successor submitted while
// this older reply was retired. Its boundary cannot revoke that input.
if (
@@ -1158,8 +1237,15 @@ export class GeminiSessionController {
content?.outputTranscription ||
content?.generationComplete ||
message.toolCall
)
) {
// Once successor output is observed, an unattributed later interruption
// cannot still claim the unobserved automatic predecessor's boundary.
if (this.typedIntent) {
this.typedIntent.automaticPredecessor = null;
this.typedIntent.retiredAutomaticPredecessor = null;
}
this.replyPending = true;
}
if (content?.outputTranscription?.text) {
this.userTurnPending = false;
this.postToolOutput = true;
@@ -1184,6 +1270,10 @@ export class GeminiSessionController {
if (message.toolCall?.functionCalls?.length)
this.enqueueTools(message.toolCall.functionCalls, epoch);
if (content?.turnComplete) {
if (this.typedIntent) {
this.typedIntent.automaticPredecessor = null;
this.typedIntent.retiredAutomaticPredecessor = null;
}
this.pttPlaybackSuppressed = false;
// A native terminal boundary also completes a silent user turn. Output,
// pending tools and queued controls still gate presentation completion.
@@ -1552,7 +1642,21 @@ export class GeminiSessionController {
throw new Error('Gemini voice is not connected');
if (this.input.pushToTalkMode && this.captureOwner)
this.captureOwner.retired = true;
this.interrupt('user-text');
const automaticPredecessor =
this.automaticTurnOpen &&
!this.replyPending &&
!this.retiredReplyPending &&
!this.deferredInput
? this.automaticActivityOwner
: null;
if (!this.interrupt('user-text')) return false;
const epoch = this.epoch;
const turn = this.turn;
const generation = this.inputGeneration;
const owned = () =>
this.owns(epoch) &&
this.turn === turn &&
this.inputGeneration === generation;
// Typed intent supersedes the held speech even if its worklet end is late.
// Close the forwarded manual activity before submitting the new text.
if (this.input.pushToTalkMode && this.forwardedActivityOpen)
@@ -1560,18 +1664,23 @@ export class GeminiSessionController {
{ realtimeInput: { activityEnd: {} } },
{ captureId: this.forwardedCaptureId },
);
if (!owned()) return false;
this.userTurns.text(clean);
if (!owned()) return false;
const typedIntent = {
epoch: this.epoch,
turn: this.turn,
responseId: `gemini-response-${this.epoch}-${++this.intentSequence}`,
automaticPredecessor,
retiredAutomaticPredecessor: null,
};
// Bind the accepted text now, rather than whatever input happens to be
// current when reliable transport or a retired reply eventually drains.
this.userTurns.expectResponse();
this.userTurns.bindResponse(typedIntent.responseId);
this.typedIntent = typedIntent;
this.beginPointerTurn('text');
this.beginPointerTurn('text', undefined, owned);
if (!owned()) return false;
const context = this.pendingPointerContext;
const mapSnapshot = [...this.pendingMapEvents];
const parts = context ? [{ text: JSON.stringify(context.item) }] : [];
@@ -1590,18 +1699,19 @@ export class GeminiSessionController {
undefined,
{
onSent: () => {
if (this.typedIntent !== typedIntent) return;
if (!owned() || this.typedIntent !== typedIntent) return;
this.replyPending = true;
this.bindTextIntent(typedIntent);
if (!owned() || this.typedIntent !== typedIntent) return;
this.consumePointerContext(context);
this.consumeMapEvents(mapSnapshot);
},
},
);
if (sent) {
if (sent && owned()) {
this.narration.userTurnEnded();
}
return sent;
return sent && owned();
}
consumeMapEvents(snapshot) {
@@ -1801,8 +1911,8 @@ export class GeminiSessionController {
}
/** Use the common pointer owner; held speech context precedes audio quietly. */
beginPointerTurn(at, capturedSnapshot = undefined) {
if (!this.pointer) return null;
beginPointerTurn(at, capturedSnapshot = undefined, isCurrent = () => true) {
if (!this.pointer || !isCurrent()) return null;
// The delayed manual speech notification still refers to the released hold.
if (
capturedSnapshot === undefined &&
@@ -1826,6 +1936,7 @@ export class GeminiSessionController {
} catch {
snapshot = null;
}
if (!isCurrent()) return null;
const chip = pointerChip(snapshot);
this.pointerAnnounced = gesture && chip ? snapshot : null;
this.emit({
@@ -1833,6 +1944,7 @@ export class GeminiSessionController {
pointer:
gesture && chip ? { ...chip, screenPx: snapshot.screenPx } : null,
});
if (!isCurrent()) return null;
const item = !chip
? this.pointerSentKey && this.pointerSentKey !== 'none'
? { type: 'pointer_context', at, target: 'none' }
@@ -1931,7 +2043,16 @@ export class GeminiSessionController {
this.intentCaptureId = captureId;
// A physical speech onset clears display-only recent captions, including
// when retirement already fenced the reply and no second interrupt runs.
const epoch = this.epoch;
const turn = this.turn;
const generation = this.inputGeneration;
this.emit({ type: 'user-caption-reset' });
if (
!this.owns(epoch) ||
this.turn !== turn ||
this.inputGeneration !== generation
)
return;
this.emit({ type: 'user-turn-start' });
}
@@ -2104,6 +2225,7 @@ export class GeminiSessionController {
} = {}) {
if (this.disposed) return;
this.epoch++;
this.inputGeneration++;
this.userTurns.reset();
this.clearSettledSpeechIntent();
this.clearControls();
+38 -15
View File
@@ -1334,21 +1334,34 @@ async function beginTestToolResponse(controller, responseId) {
}
async function continueTestToolResponse(controller, previousId, responseId) {
await controller.handleRealtimeEvent({
data: JSON.stringify({
type: 'response.done',
response: {
id: previousId,
status: 'completed',
output: [{ type: 'function_call' }],
},
}),
});
controller.queueResponseCreate('Continue the current user request.');
let metadata;
const channel = controller.dc;
const send = channel.send;
channel.send = function (body) {
const payload = JSON.parse(body);
if (payload.type === 'response.create')
metadata = payload.response?.metadata;
return send.call(this, body);
};
try {
await controller.handleRealtimeEvent({
data: JSON.stringify({
type: 'response.done',
response: {
id: previousId,
status: 'completed',
output: [{ type: 'function_call' }],
},
}),
});
controller.queueResponseCreate('Continue the current user request.');
} finally {
channel.send = send;
}
await controller.handleRealtimeEvent({
data: JSON.stringify({
type: 'response.created',
response: { id: responseId },
response: { id: responseId, metadata },
}),
});
}
@@ -1437,7 +1450,11 @@ test('Radio handoff waits for response.done so later multi-intent tools execute
await controller.handleRealtimeEvent(
event({
type: 'response.done',
response: { id: 'response-1', status: 'completed' },
response: {
id: 'response-1',
status: 'completed',
output: [{ type: 'function_call' }],
},
}),
);
assert.deepEqual(order, [
@@ -1610,7 +1627,10 @@ test('detected speech cancels a prepared Radio handoff before a cancelled respon
);
await controller.handleRealtimeEvent(
event({ type: 'input_audio_buffer.speech_started' }),
event({
type: 'input_audio_buffer.speech_started',
item_id: 'speech-radio-confirmation',
}),
);
assert.equal(controller.pendingRadioPlaybackResult, null);
await controller.handleRealtimeEvent(
@@ -1681,7 +1701,10 @@ test('a Radio tool result that resolves after speech interruption cannot re-arm
await Promise.resolve();
assert.equal(receivedSignal?.aborted, false);
await controller.handleRealtimeEvent(
event({ type: 'input_audio_buffer.speech_started' }),
event({
type: 'input_audio_buffer.speech_started',
item_id: 'speech-radio-in-flight',
}),
);
assert.equal(
receivedSignal.aborted,
+542
View File
@@ -0,0 +1,542 @@
import test from 'node:test';
import assert from 'node:assert/strict';
import { GevRealtimeController } from './realtimeController.js';
import {
readVoiceUserTurnText,
subscribeVoiceUserTurn,
} from './userTurnIntent.js';
function fixture({ onEvent, pending = false } = {}) {
const sent = [],
calls = [],
events = [],
releases = [];
const effects = { metrics: 0, speech: 0, response: 0 };
let c;
c = new GevRealtimeController({
runner: async (_name, _args, options) => {
calls.push(options);
return pending
? new Promise((resolve) => releases.push(resolve))
: { ok: true };
},
backend: {},
debugSink: null,
onSessionEvent: (event) => {
events.push(event);
onEvent?.(event, c);
},
ui: {
root: {
dataset: {},
classList: { remove() {} },
querySelectorAll: () => [],
},
status: {},
detail: {},
},
});
const channel = () => ({
readyState: 'open',
send: (v) => sent.push(JSON.parse(v)),
close() {
this.readyState = 'closed';
},
});
c.dc = channel();
c.status = 'listening';
c._turns.narration = null;
for (const [owner, name, effect] of [
[c._metrics, 'serverEvent', 'metrics'],
[c._turns, 'observeSpeech', 'speech'],
[c._turns, 'updateResponseState', 'response'],
]) {
const original = owner[name].bind(owner);
owner[name] = (...args) => {
effects[effect]++;
return original(...args);
};
}
const event = (payload, source = c.dc) =>
c.handleRealtimeEvent({
data: JSON.stringify(payload),
currentTarget: source,
});
const created = (id, metadata) =>
event({ type: 'response.created', response: { id, metadata } });
const done = (id, tool = false) =>
event({
type: 'response.done',
response: {
id,
status: 'completed',
output: tool ? [{ type: 'function_call' }] : [],
},
});
const tool = (id, call) =>
event({
type: 'response.function_call_arguments.done',
response_id: id,
call_id: call,
name: 'get_entity_context',
arguments: '{}',
});
const speech = async (id, text) => {
await event({ type: 'input_audio_buffer.speech_started', item_id: id });
await event({
type: 'conversation.item.input_audio_transcription.completed',
item_id: id,
transcript: text,
});
await event({ type: 'input_audio_buffer.committed', item_id: id });
};
const request = () => sent.findLast((e) => e.type === 'response.create');
const output = () =>
sent
.filter((e) => e.item?.type === 'function_call_output')
.map((e) => JSON.parse(e.item.output));
const snapshot = () => ({
effects: { ...effects },
events: events.length,
pending: c._turns.userTurnIntent.reservationState(),
active: c.activeResponseId,
responseActive: c.responseActive,
responsePending: c.responseCreatePending,
typedPending: c.pendingUserTextResponse,
instructions: c.pendingResponseInstructions,
});
const close = () => {
releases.splice(0).forEach((r) => r({ ok: true }));
c.stop();
};
return {
c,
sent,
calls,
events,
effects,
channel,
event,
created,
done,
tool,
speech,
request,
output,
snapshot,
close,
};
}
for (const code of [
'conversation_already_has_active_response',
'server_error',
]) {
test(`OpenAI ${code} retirement cannot erase the token observer's newer typed request`, async () => {
const f = fixture();
try {
f.c.sendTextCommand('original T0');
await f.created('T0', f.request().response.metadata);
const original = f.c._turns.userTurnIntent.forResponse('T0');
await f.tool('T0', 'T0-tool');
await f.done('T0', true);
const rejected = f.request();
let armed = true;
subscribeVoiceUserTurn(original, (token) => {
if (armed && !readVoiceUserTurnText(token)) {
armed = false;
f.c.sendTextCommand('observer T1');
}
});
await f.event({
type: 'error',
error: {
code,
event_id: rejected.event_id,
message: 'synthetic rejection',
},
});
assert.equal(readVoiceUserTurnText(original), undefined);
if (code === 'conversation_already_has_active_response') {
assert.equal(f.c.pendingUserTextResponse, true);
await f.done('other-active-response');
}
const successor = f.request();
assert.notEqual(
successor.event_id,
rejected.event_id,
'new input has its own response request',
);
await f.created('T1', successor.response.metadata);
await f.tool('T1', 'T1-tool');
assert.equal(
readVoiceUserTurnText(f.calls.at(-1).userTurn),
'observer T1',
);
assert.equal(
f.sent.filter(
(e) =>
e.type === 'response.create' && e.event_id === rejected.event_id,
).length,
1,
'rejected request is not retried',
);
} finally {
f.close();
}
});
}
for (const count of [1, 20]) {
test(`OpenAI completed response replay after ${count} input records cannot borrow newer speech authority`, async () => {
const f = fixture();
try {
for (let i = 0; i < count; i++) {
await f.speech(`old-${i}`, `old speech ${i}`);
await f.created(`old-response-${i}`);
await f.done(`old-response-${i}`);
}
await f.speech('C', 'genuine C');
const before = f.snapshot();
await f.created('old-response-0');
assert.deepEqual(
f.snapshot(),
before,
'duplicate is rejected before every response semantic consumer',
);
await f.tool('old-response-0', 'replayed-tool');
assert.equal(f.calls.length, 0);
assert.equal(f.c._turns.userTurnIntent.reservationState().pending, 1);
await f.created('genuine-C');
await f.tool('genuine-C', 'C-tool');
assert.equal(readVoiceUserTurnText(f.calls[0].userTurn), 'genuine C');
assert.equal(f.c._turns.userTurnIntent.reservationState().pending, 0);
} finally {
f.close();
}
});
}
for (const code of [
'conversation_already_has_active_response',
'server_error',
]) {
test(`OpenAI ${code} cleanup preserves input submitted synchronously by tool cancellation`, async () => {
const f = fixture({ pending: true });
let action;
try {
f.c.sendTextCommand('original T0');
await f.created('T0', f.request().response.metadata);
action = f.tool('T0', 'pending-T0');
await new Promise((resolve) => setImmediate(resolve));
assert.equal(f.calls.length, 1);
await f.done('T0', true);
f.c.queueResponseCreate('Continue original T0.');
const rejected = f.request();
f.calls[0].signal.addEventListener(
'abort',
() => f.c.sendTextCommand('cancel observer T1'),
{ once: true },
);
await f.event({
type: 'error',
error: {
code,
event_id: rejected.event_id,
message: 'synthetic rejection',
},
});
assert.equal(f.calls[0].signal.aborted, true);
if (code === 'conversation_already_has_active_response') {
assert.equal(f.c.pendingUserTextResponse, true);
await f.done('other-active');
}
const next = f.request();
assert.notEqual(next.event_id, rejected.event_id);
await f.created('T1', next.response.metadata);
assert.equal(
readVoiceUserTurnText(f.c._turns.userTurnIntent.forResponse('T1')),
'cancel observer T1',
);
} finally {
f.close();
if (action) await action;
}
});
}
for (const code of ['conversation_already_has_active_response', 'server_error'])
for (const replace of [false, true]) {
test(`OpenAI ${code} token retirement ${replace ? 'replacement' : 'Stop'} cannot resume old error cleanup`, async () => {
const f = fixture();
try {
f.c.sendTextCommand('original T0');
await f.created('T0', f.request().response.metadata);
const token = f.c._turns.userTurnIntent.forResponse('T0');
await f.tool('T0', 'T0-tool');
await f.done('T0', true);
const rejected = f.request();
let armed = true;
subscribeVoiceUserTurn(token, (value) => {
if (armed && !readVoiceUserTurnText(value)) {
armed = false;
f.c.stop();
if (replace) {
f.c.dc = f.channel();
f.c.status = 'listening';
f.c.sendTextCommand('replacement T1');
}
}
});
await f.event({
type: 'error',
error: {
code,
event_id: rejected.event_id,
message: 'synthetic rejection',
},
});
if (replace) {
const next = f.request();
await f.created('replacement', next.response.metadata);
await f.tool('replacement', 'replacement-tool');
assert.equal(
readVoiceUserTurnText(f.calls.at(-1).userTurn),
'replacement T1',
);
} else {
assert.equal(f.c.dc, null);
assert.equal(f.c.status, 'idle');
assert.equal(f.c._turns.userTurnIntent.reservationState().pending, 0);
}
} finally {
f.close();
}
});
}
for (const id of [undefined, null, '', 8, {}, []]) {
test(`OpenAI malformed response-created ID ${JSON.stringify(id)} is inert`, async () => {
const f = fixture();
try {
await f.speech('C', 'genuine C');
const before = f.snapshot();
await f.created(id);
assert.deepEqual(f.snapshot(), before);
await f.created('valid');
await f.tool('valid', 'valid-tool');
assert.equal(readVoiceUserTurnText(f.calls[0].userTurn), 'genuine C');
} finally {
f.close();
}
});
}
test('OpenAI duplicate current response-created preserves its running tool and distinct subsequent events', async () => {
const f = fixture({ pending: true });
let action;
try {
await f.speech('B', 'genuine B');
await f.created('B-response');
const token = f.c._turns.userTurnIntent.forResponse('B-response');
action = f.tool('B-response', 'first-tool');
await new Promise((resolve) => setImmediate(resolve));
const before = f.snapshot();
await f.created('B-response');
assert.deepEqual(f.snapshot(), before);
assert.equal(f.calls[0].signal.aborted, false);
assert.equal(readVoiceUserTurnText(token), 'genuine B');
await f.event({
type: 'response.output_audio_transcript.done',
response_id: 'B-response',
transcript: 'normal distinct transcript',
});
assert.ok(
f.effects.speech > before.effects.speech,
'later distinct response events are not deduplicated',
);
} finally {
f.close();
if (action) await action;
}
});
test('OpenAI wrong or closed transport cannot admit/poison a response identity', async () => {
const f = fixture();
try {
await f.speech('B', 'genuine B');
const before = f.snapshot();
await f.event(
{ type: 'response.created', response: { id: 'reusable' } },
f.channel(),
);
assert.deepEqual(f.snapshot(), before);
f.c.dc.readyState = 'closed';
await f.created('reusable');
assert.deepEqual(f.snapshot(), before);
f.c.dc.readyState = 'open';
await f.created('reusable');
await f.tool('reusable', 'valid-tool');
assert.equal(readVoiceUserTurnText(f.calls[0].userTurn), 'genuine B');
} finally {
f.close();
}
});
for (const progress of [false, true]) {
test(`OpenAI ${progress ? 'progress' : 'out-of-band'} response replay is rejected before consumers`, async () => {
const f = fixture();
try {
const payload = {
type: 'response.created',
response: {
id: 'outside',
conversation_id: null,
metadata: progress ? { gev: 'progress' } : { topic: 'other' },
},
};
await f.event(payload);
const before = f.snapshot();
await f.event(payload);
assert.deepEqual(f.snapshot(), before);
await f.event({
type: 'response.done',
response: {
id: 'outside',
status: 'completed',
metadata: payload.response.metadata,
output: [],
},
});
assert.ok(
f.effects.metrics > before.effects.metrics,
'distinct done is processed',
);
} finally {
f.close();
}
});
}
test('OpenAI Stop/reset releases response history and legitimate new connection may reuse the same response ID', async () => {
const f = fixture();
try {
await f.speech('B', 'old speech');
await f.created('same-response');
await f.done('same-response');
const oldChannel = f.c.dc;
const oldHistory = f.c._turns.admittedResponseIds;
f.c.stop();
assert.equal(oldHistory.size, 0);
assert.notEqual(f.c._turns.admittedResponseIds, oldHistory);
f.c.dc = f.channel();
f.c.status = 'listening';
await f.speech('B', 'new speech');
const before = f.snapshot();
await f.event(
{ type: 'response.created', response: { id: 'same-response' } },
oldChannel,
);
assert.deepEqual(f.snapshot(), before);
await f.created('same-response');
await f.tool('same-response', 'new-tool');
assert.equal(readVoiceUserTurnText(f.calls[0].userTurn), 'new speech');
} finally {
f.close();
}
});
for (const via of ['boundary', 'metrics', 'speaker']) {
test(`OpenAI accepted response callback ${via} Stop cannot resurrect old response state`, async () => {
let armed = false;
const f = fixture({
onEvent: (event, c) => {
if (armed && via === 'boundary' && event.type === 'response-start') {
armed = false;
c.stop();
}
},
});
try {
if (via === 'metrics') {
const original = f.c._metrics.serverEvent.bind(f.c._metrics);
f.c._metrics.serverEvent = (payload) => {
original(payload);
if (armed && payload.type === 'response.created') {
armed = false;
f.c.stop();
}
};
}
if (via === 'speaker') {
const original = f.c._turns.setVoiceSpeaker;
f.c._turns.setVoiceSpeaker = (...args) => {
original(...args);
if (armed) {
armed = false;
f.c.stop();
}
};
}
await f.speech('B', 'old speech');
armed = true;
await f.created('old-response');
assert.equal(f.c.dc, null);
assert.equal(f.c.status, 'idle');
assert.equal(f.c.responseActive, false);
assert.equal(f.c.activeResponseId, null);
assert.equal(f.c._turns.admittedResponseIds.size, 0);
} finally {
f.close();
}
});
}
test('OpenAI accepted response keeps distinct sibling tool calls after a duplicate creation', async () => {
const f = fixture();
try {
await f.speech('B', 'genuine B');
await f.created('B-response');
const token = f.c._turns.userTurnIntent.forResponse('B-response');
await f.tool('B-response', 'first');
await f.created('B-response');
await f.tool('B-response', 'second');
assert.equal(f.calls.length, 2);
assert.equal(f.calls[0].userTurn, token);
assert.equal(f.calls[1].userTurn, token);
} finally {
f.close();
}
});
test('OpenAI nested replacement can reuse a response ID without the old accepted event resuming into it', async () => {
let armed = false;
let nested;
let f;
f = fixture({
onEvent: (event, c) => {
if (armed && event.type === 'response-start') {
armed = false;
c.stop();
c.dc = f.channel();
c.status = 'listening';
c.sendTextCommand('replacement typed input');
nested = f.created('shared-response', f.request().response.metadata);
}
},
});
try {
await f.speech('B', 'old B');
armed = true;
await f.created('shared-response');
await nested;
assert.equal(f.c.responseActive, true);
assert.equal(f.c.activeResponseId, 'shared-response');
const token = f.c._turns.userTurnIntent.forResponse('shared-response');
assert.equal(readVoiceUserTurnText(token), 'replacement typed input');
await f.tool('shared-response', 'replacement-tool');
assert.equal(f.calls[0].userTurn, token);
} finally {
f.close();
}
});
@@ -0,0 +1,474 @@
import test from 'node:test';
import assert from 'node:assert/strict';
import { GevRealtimeController } from './realtimeController.js';
import {
readVoiceUserTurnText,
subscribeVoiceUserTurn,
} from './userTurnIntent.js';
function fixture({ via, onEvent, send } = {}) {
const sent = [],
calls = [],
events = [];
let armed = false;
const c = new GevRealtimeController({
runner: async (_name, _args, options) => {
calls.push(options.userTurn);
return { ok: true };
},
backend: {},
debugSink: null,
onSessionEvent: (event) => {
events.push(event);
if (armed && via === 'boundary' && event.type === 'user-turn-start') {
armed = false;
c.sendTextCommand('newer typed request');
}
onEvent?.(event, c);
},
ui: {
root: {
dataset: {},
classList: { remove() {} },
querySelectorAll: () => [],
},
status: {},
detail: {},
},
});
const channel = () => ({
readyState: 'open',
send: (value) => {
const payload = JSON.parse(value);
sent.push(payload);
send?.(payload, c);
},
close() {
this.readyState = 'closed';
},
});
c.dc = channel();
c.status = 'listening';
c._turns.narration = null;
const event = (payload) =>
c.handleRealtimeEvent({
data: JSON.stringify(payload),
currentTarget: c.dc,
});
const start = (id) =>
event({ type: 'input_audio_buffer.speech_started', item_id: id });
const transcript = (id, text) =>
event({
type: 'conversation.item.input_audio_transcription.completed',
item_id: id,
transcript: text,
});
const commit = (id) =>
event({ type: 'input_audio_buffer.committed', item_id: id });
const created = (id, metadata) =>
event({
type: 'response.created',
response: { id, ...(metadata === undefined ? {} : { metadata }) },
});
const done = (id, tool = false) =>
event({
type: 'response.done',
response: {
id,
status: 'completed',
output: tool ? [{ type: 'function_call' }] : [],
},
});
const tool = (id, call) =>
event({
type: 'response.function_call_arguments.done',
response_id: id,
call_id: call,
name: 'get_entity_context',
arguments: '{}',
});
const request = () =>
sent.findLast(
(e) =>
e.type === 'response.create' &&
e.response?.metadata?.gev !== 'progress',
);
const output = () =>
sent
.filter((e) => e.item?.type === 'function_call_output')
.map((e) => JSON.parse(e.item.output));
const prepare = async () => {
await start('B');
await transcript('B', 'old B');
await commit('B');
await created('R0');
if (via === 'token')
subscribeVoiceUserTurn(
c._turns.userTurnIntent.forResponse('R0'),
(token) => {
if (armed && !readVoiceUserTurnText(token)) {
armed = false;
c.sendTextCommand('newer typed request');
}
},
);
armed = true;
await start('C');
await transcript('C', 'older C speech');
};
return {
c,
sent,
calls,
events,
channel,
event,
start,
transcript,
commit,
created,
done,
tool,
request,
output,
prepare,
close: () => c.stop(),
};
}
for (const via of ['boundary', 'token'])
for (const commitFirst of [false, true])
for (const typedFirst of [false, true]) {
test(`OpenAI correlation ${via}: commit ${commitFirst ? 'before' : 'after'} old done, ${typedFirst ? 'typed' : 'speech'} response first`, async () => {
const f = fixture({ via });
try {
await f.prepare();
assert.equal(f.c._turns.userTurnIntent.reservationState().pending, 0);
if (commitFirst) await f.commit('C');
await f.done('R0');
const request = f.request();
if (!commitFirst) await f.commit('C');
if (typedFirst) {
await f.created('typed-T', request?.response?.metadata);
await f.created('automatic-C');
} else {
await f.created('automatic-C');
await f.created('typed-T', request?.response?.metadata);
}
const typed = f.c._turns.userTurnIntent.forResponse('typed-T');
await f.tool('automatic-C', 'C-call');
assert.equal(
f.calls.length,
0,
'spoken C must never run with typed T authority',
);
assert.equal(f.output().at(-1).superseded, true);
assert.equal(f.c.pendingResponseInstructions, null);
assert.equal(readVoiceUserTurnText(typed), 'newer typed request');
await f.tool('typed-T', 'T-call');
assert.equal(f.calls.length, 1);
assert.equal(f.calls[0], typed);
assert.equal(
readVoiceUserTurnText(f.calls[0]),
'newer typed request',
);
assert.equal(
request.response.metadata.gev_request_id,
request.event_id,
);
} finally {
f.close();
}
});
}
for (const tag of ['unknown', '', 7, {}, [], null]) {
test(`OpenAI unknown or malformed client correlation ${JSON.stringify(tag)} cannot consume speech or typed reservations`, async () => {
const f = fixture();
try {
await f.start('B');
await f.transcript('B', 'older speech');
await f.commit('B');
f.c.sendTextCommand('newer typed request');
const request = f.request();
assert.equal(f.c._turns.userTurnIntent.reservationState().pending, 2);
await f.created('foreign', { gev_request_id: tag });
await f.tool('foreign', 'foreign-tool');
assert.equal(f.calls.length, 0);
assert.equal(f.output().at(-1).superseded, true);
assert.equal(f.c.pendingResponseInstructions, null);
assert.equal(f.c._turns.userTurnIntent.reservationState().pending, 2);
await f.created('automatic-B');
assert.equal(f.c._turns.isSupersededResponse('automatic-B'), true);
await f.created('typed-T', request.response.metadata);
await f.tool('typed-T', 'typed-tool');
assert.equal(readVoiceUserTurnText(f.calls[0]), 'newer typed request');
} finally {
f.close();
}
});
}
test('OpenAI absent client metadata cannot borrow typed authority or pass it through an unverified followup', async () => {
const f = fixture();
try {
f.c.sendTextCommand('newer typed request');
const request = f.request();
await f.created('unrelated-automatic');
await f.tool('unrelated-automatic', 'unverified-call');
assert.equal(f.calls.length, 0);
assert.equal(f.output().at(-1).voiceIntentUnverified, true);
assert.equal(
f.c.pendingResponseInstructions,
null,
'untracked response cannot queue a followup with current typed authority',
);
assert.equal(f.c._turns.userTurnIntent.reservationState().pending, 1);
await f.created('typed-T', request.response.metadata);
await f.tool('typed-T', 'real-T');
assert.equal(readVoiceUserTurnText(f.calls[0]), 'newer typed request');
} finally {
f.close();
}
});
test('OpenAI tagged followup keeps the exact originating request and ignores out-of-band progress', async () => {
const f = fixture();
try {
f.c.sendTextCommand('newer typed request');
await f.created('typed-T', f.request().response.metadata);
const token = f.c._turns.userTurnIntent.forResponse('typed-T');
await f.tool('typed-T', 'T-tool');
await f.done('typed-T', true);
const followup = f.request();
assert.ok(followup.response.instructions);
assert.equal(followup.response.metadata.gev_request_id, followup.event_id);
await f.created('progress', { gev: 'progress' });
assert.equal(f.c._turns.userTurnIntent.reservationState().pending, 1);
await f.created('followup', followup.response.metadata);
assert.equal(f.c._turns.userTurnIntent.forResponse('followup'), token);
await f.done('followup');
assert.equal(readVoiceUserTurnText(token), undefined);
} finally {
f.close();
}
});
test('OpenAI duplicate tagged response events do not consume another request or speech slot', async () => {
const f = fixture();
try {
f.c.sendTextCommand('newer typed request');
const metadata = f.request().response.metadata;
await f.created('typed-T', metadata);
const token = f.c._turns.userTurnIntent.forResponse('typed-T');
await f.created('typed-T', metadata);
assert.equal(f.c._turns.userTurnIntent.forResponse('typed-T'), token);
await f.start('C');
await f.transcript('C', 'new speech');
await f.commit('C');
await f.created('duplicate-request-response', metadata);
assert.equal(
f.c._turns.isSupersededResponse('duplicate-request-response'),
true,
);
assert.equal(f.c._turns.userTurnIntent.reservationState().pending, 1);
await f.created('C-response');
await f.tool('C-response', 'C-tool');
assert.equal(readVoiceUserTurnText(f.calls[0]), 'new speech');
} finally {
f.close();
}
});
for (const lifecycle of ['stop', 'replace', 'new typed']) {
test(`OpenAI synchronous ${lifecycle} during actual response send cannot reserve newer authority`, async () => {
let armed = true;
let sentRequest;
let f;
f = fixture({
send: (payload, c) => {
if (!armed || payload.type !== 'response.create') return;
armed = false;
sentRequest = payload;
if (lifecycle === 'new typed') c.sendTextCommand('new typed successor');
else {
c.stop();
if (lifecycle === 'replace') {
c.dc = f.channel();
c.status = 'listening';
}
}
},
});
try {
f.c.sendTextCommand('original typed request');
if (lifecycle === 'stop') {
assert.equal(f.c.dc, null);
assert.equal(f.c._turns.userTurnIntent.reservationState().pending, 0);
} else {
if (lifecycle === 'replace') {
assert.equal(f.c._turns.userTurnIntent.reservationState().pending, 0);
f.c.sendTextCommand('new typed successor');
}
await f.created('old-sent-response', sentRequest.response.metadata);
await f.tool('old-sent-response', 'old-tool');
assert.equal(f.calls.length, 0);
assert.equal(f.output().at(-1).superseded, true);
if (lifecycle === 'new typed') await f.done('old-sent-response');
const latest = f.request();
assert.notEqual(latest.event_id, sentRequest.event_id);
await f.created('new-response', latest.response.metadata);
await f.tool('new-response', 'new-tool');
assert.equal(readVoiceUserTurnText(f.calls[0]), 'new typed successor');
}
} finally {
f.close();
}
});
}
for (const stage of ['conversation.item.create', 'response.create']) {
test(`OpenAI failed ${stage} never reserves a response or authorizes an automatic reply`, async () => {
const f = fixture({
send: (payload) => {
if (payload.type === stage) throw new Error('synthetic send failure');
},
});
try {
f.c.sendTextCommand('newer typed request');
assert.equal(f.c._turns.userTurnIntent.reservationState().pending, 0);
await f.created('untracked');
await f.tool('untracked', 'untracked-tool');
assert.equal(f.calls.length, 0);
} finally {
f.close();
}
});
}
test('OpenAI rejected tagged request cannot consume later genuine speech authority', async () => {
const f = fixture();
try {
f.c.sendTextCommand('old typed request');
const rejected = f.request();
await f.start('C');
await f.transcript('C', 'genuine speech');
await f.commit('C');
await f.event({
type: 'error',
event_id: 'server-error',
error: {
code: 'conversation_already_has_active_response',
event_id: rejected.event_id,
message: 'synthetic overlap',
},
});
await f.created('rejected-request-response', rejected.response.metadata);
assert.equal(
f.c._turns.isSupersededResponse('rejected-request-response'),
true,
);
await f.created('C-response');
await f.tool('C-response', 'C-tool');
assert.equal(readVoiceUserTurnText(f.calls[0]), 'genuine speech');
} finally {
f.close();
}
});
test('OpenAI local barge-in drops a deferred typed capture before a late old completion', async () => {
const f = fixture();
try {
await f.created('R0');
f.c.sendTextCommand('deferred typed');
assert.equal(f.c.pendingUserTextResponse, true);
f.c._turns.bargeIn();
await f.done('R0');
assert.equal(f.sent.filter((e) => e.type === 'response.create').length, 0);
assert.equal(f.c.pendingUserTextResponse, false);
assert.equal(f.c._turns.userTurnIntent.reservationState().pending, 0);
} finally {
f.close();
}
});
test('OpenAI unrelated explicit out-of-band response never consumes a tagged typed request', async () => {
const f = fixture();
try {
f.c.sendTextCommand('newer typed request');
const request = f.request();
await f.event({
type: 'response.created',
response: {
id: 'foreign-oob',
conversation_id: null,
metadata: request.response.metadata,
},
});
await f.tool('foreign-oob', 'oob-tool');
assert.equal(f.calls.length, 0);
assert.equal(f.c.pendingResponseInstructions, null);
assert.equal(f.c._turns.userTurnIntent.reservationState().pending, 1);
await f.created('typed-T', request.response.metadata);
await f.tool('typed-T', 'T-tool');
assert.equal(readVoiceUserTurnText(f.calls[0]), 'newer typed request');
} finally {
f.close();
}
});
test('OpenAI duplicate late commits for retained retired C do not duplicate reservations', async () => {
const f = fixture({ via: 'boundary' });
try {
await f.prepare();
await f.done('R0');
const request = f.request();
await f.commit('C');
await f.commit('C');
assert.equal(f.c._turns.userTurnIntent.reservationState().pending, 2);
await f.created('automatic-C');
await f.commit('C');
assert.equal(f.c._turns.userTurnIntent.reservationState().pending, 1);
await f.created('typed-T', request.response.metadata);
await f.tool('typed-T', 'T-tool');
assert.equal(readVoiceUserTurnText(f.calls[0]), 'newer typed request');
assert.equal(f.c._turns.userTurnIntent.reservationState().pending, 0);
} finally {
f.close();
}
});
test('OpenAI a newer typed input during a followup send cannot own the old followup response', async () => {
let armed = true;
const f = fixture({
send: (payload, c) => {
if (
armed &&
payload.type === 'response.create' &&
payload.response?.instructions
) {
armed = false;
c.sendTextCommand('new typed successor');
}
},
});
try {
f.c.sendTextCommand('original typed request');
await f.created('T-original', f.request().response.metadata);
await f.tool('T-original', 'original-tool');
await f.done('T-original', true);
const oldFollowup = f.request();
await f.created('old-followup', oldFollowup.response.metadata);
await f.tool('old-followup', 'old-followup-tool');
assert.equal(f.calls.length, 1);
assert.equal(f.output().at(-1).superseded, true);
await f.done('old-followup');
const latest = f.request();
assert.notEqual(latest.event_id, oldFollowup.event_id);
await f.created('T-successor', latest.response.metadata);
await f.tool('T-successor', 'successor-tool');
assert.equal(readVoiceUserTurnText(f.calls[1]), 'new typed successor');
} finally {
f.close();
}
});
+734
View File
@@ -0,0 +1,734 @@
// Provider-item admission must precede every semantic speech-start consumer.
import test from 'node:test';
import assert from 'node:assert/strict';
import { GevRealtimeController } from './realtimeController.js';
import {
readVoiceUserTurnText,
readVoiceUserTurnProvenance,
subscribeVoiceUserTurn,
} from './userTurnIntent.js';
const tick = () => new Promise((resolve) => setImmediate(resolve));
function fixture({ pending = false, onEvent } = {}) {
const events = [],
sent = [],
calls = [],
releases = [];
const effects = { narration: 0, radio: 0, pointer: 0, metrics: 0 };
let c;
c = new GevRealtimeController({
runner: async (name, args, options) => {
calls.push({ name, args, options });
return pending
? new Promise((resolve) => releases.push(resolve))
: { ok: true };
},
backend: {},
debugSink: null,
onSessionEvent: (event) => {
events.push(event);
onEvent?.(event, c);
},
ui: {
root: {
dataset: {},
classList: { remove() {} },
querySelectorAll: () => [],
},
status: {},
detail: {},
},
});
const channel = () => ({
readyState: 'open',
send: (value) => sent.push(JSON.parse(value)),
close() {
this.readyState = 'closed';
},
});
c.dc = channel();
c.status = 'listening';
const wrap = (object, name, key) => {
const original = object[name].bind(object);
object[name] = (...args) => {
effects[key]++;
return original(...args);
};
};
wrap(c._turns.narration, 'cancel', 'narration');
wrap(c, 'cancelRadioHandoff', 'radio');
wrap(c, 'beginPointerTurn', 'pointer');
wrap(c._metrics, 'serverEvent', 'metrics');
const event = (payload, source = c.dc) =>
c.handleRealtimeEvent({
data: JSON.stringify(payload),
currentTarget: source,
});
const start = (id) =>
event({ type: 'input_audio_buffer.speech_started', item_id: id });
const transcript = (id, text) =>
event({
type: 'conversation.item.input_audio_transcription.completed',
item_id: id,
transcript: text,
});
const commit = (id) =>
event({ type: 'input_audio_buffer.committed', item_id: id });
const created = (id, metadata) =>
event({
type: 'response.created',
response: { id, ...(metadata ? { metadata } : {}) },
});
const done = (id) =>
event({
type: 'response.done',
response: { id, status: 'completed', output: [] },
});
const tool = (id, callId) =>
event({
type: 'response.function_call_arguments.done',
response_id: id,
call_id: callId,
name: 'adjust_camera_zoom',
arguments: '{"direction":"in","amount":"little"}',
});
const live = async (mode = 'completed') => {
if (mode === 'disabled') c._turns.bindTranscriptionModel('off');
await start('B');
if (mode === 'completed') await transcript('B', 'zoom in a little');
if (mode === 'failed')
await event({
type: 'conversation.item.input_audio_transcription.failed',
item_id: 'B',
});
await commit('B');
await created('B-response');
const action = tool('B-response', 'pending-B');
await tick();
if (mode !== 'pending') assert.equal(calls.length, 1);
return action;
};
const release = () =>
releases.splice(0).forEach((resolve) => resolve({ ok: true }));
const snapshot = () => ({
effects: { ...effects },
events: events.length,
pendingTyped: c.pendingUserTextResponse,
pendingInstructions: c.pendingResponseInstructions,
pendingUser: c.userTurnPending,
reservations: c._turns.userTurnIntent.reservationState(),
});
const output = () =>
sent
.filter((x) => x.item?.type === 'function_call_output')
.map((x) => JSON.parse(x.item.output));
const close = () => {
release();
c.stop();
};
return {
c,
events,
sent,
calls,
event,
start,
transcript,
commit,
created,
clientCreated: (id) =>
created(
id,
sent.findLast(
(e) =>
e.type === 'response.create' &&
e.response?.metadata?.gev !== 'progress',
)?.response?.metadata,
),
done,
tool,
live,
release,
snapshot,
output,
channel,
close,
};
}
for (const replay of ['current', 'older', 'evicted']) {
test(`OpenAI ${replay} speech-start replay leaves token, running tool and every boundary consumer unchanged`, async () => {
const f = fixture({ pending: true });
let action;
try {
if (replay === 'older') await f.start('A');
if (replay === 'evicted')
for (let i = 0; i < 20; i++) await f.start(`mic-${i}`);
// Keep this Promise unawaited while the actual dispatcher is executing.
action = f.live();
await tick();
assert.equal(f.calls.length, 1);
const { userTurn, signal } = f.calls[0].options;
f.c.pendingResponseInstructions = 'Retain the legitimate follow-up';
const before = f.snapshot();
await f.start(
replay === 'current' ? 'B' : replay === 'older' ? 'A' : 'mic-0',
);
assert.deepEqual(f.snapshot(), before);
assert.equal(readVoiceUserTurnText(userTurn), 'zoom in a little');
assert.equal(
f.c._turns.userTurnIntent.forResponse('B-response'),
userTurn,
);
assert.equal(signal.aborted, false);
f.release();
await action;
assert.equal(f.output().at(-1).ok, true);
} finally {
f.release();
if (action) await action;
f.close();
}
});
}
for (const mode of ['disabled', 'failed']) {
test(`OpenAI duplicate retains ${mode} transcription provenance for the actual runner`, async () => {
const f = fixture({ pending: true });
let action;
try {
action = f.live(mode);
await tick();
const token = f.calls[0].options.userTurn;
const before = f.snapshot();
await f.start('B');
assert.deepEqual(f.snapshot(), before);
assert.deepEqual(readVoiceUserTurnProvenance(token), {
inputKind: 'speech',
transcriptionStatus: mode,
});
assert.equal(f.calls[0].options.signal.aborted, false);
f.release();
await action;
assert.equal(f.output().at(-1).ok, true);
} finally {
f.release();
if (action) await action;
f.close();
}
});
}
test('OpenAI duplicate cannot replace a pending-transcription waiter', async () => {
const f = fixture();
let action;
try {
await f.start('B');
await f.commit('B');
await f.created('B-response');
action = f.tool('B-response', 'waiter');
await tick();
assert.equal(f.calls.length, 0);
await f.start('B');
await f.transcript('B', 'zoom in a little');
await action;
assert.equal(f.calls.length, 1);
assert.equal(
readVoiceUserTurnText(f.calls[0].options.userTurn),
'zoom in a little',
);
assert.equal(f.output().at(-1).ok, true);
} finally {
if (action) await action;
f.close();
}
});
test('OpenAI duplicate cannot drop queued typed intent or insert a response reservation', async () => {
const f = fixture();
try {
await f.start('B');
await f.transcript('B', 'first input');
await f.commit('B');
await f.created('B-response');
f.c.sendTextCommand('zoom out typed');
assert.equal(f.c.pendingUserTextResponse, true);
const before = f.snapshot();
await f.start('B');
assert.deepEqual(f.snapshot(), before);
await f.done('B-response');
await f.clientCreated('typed-response');
const token = f.c._turns.userTurnIntent.forResponse('typed-response');
assert.equal(readVoiceUserTurnText(token), 'zoom out typed');
await f.tool('typed-response', 'typed-current');
assert.equal(f.calls.length, 1);
assert.equal(f.calls[0].options.userTurn, token);
} finally {
f.close();
}
});
test('OpenAI a genuine unseen onset still aborts the old tool and admits the new request', async () => {
const f = fixture({ pending: true });
let action;
try {
action = f.live();
await tick();
const old = f.calls[0].options;
const before = f.events.length;
await f.start('C');
assert.equal(readVoiceUserTurnText(old.userTurn), undefined);
assert.equal(old.signal.aborted, true);
assert.equal(
f.events.slice(before).filter((e) => e.type === 'user-turn-start').length,
1,
);
assert.equal(
f.events.slice(before).filter((e) => e.type === 'interruption').length,
1,
);
f.release();
await action;
assert.equal(f.output().at(-1).superseded, true);
await f.done('B-response');
await f.transcript('C', 'zoom out new input');
await f.commit('C');
await f.created('C-response');
action = f.tool('C-response', 'C-tool');
await tick();
assert.equal(
readVoiceUserTurnText(f.calls[1].options.userTurn),
'zoom out new input',
);
f.release();
await action;
} finally {
f.release();
if (action) await action;
f.close();
}
});
test('OpenAI genuine onset still drops an unsent typed response, unlike a replay', async () => {
const f = fixture();
try {
await f.start('B');
await f.commit('B');
await f.created('B-response');
f.c.sendTextCommand('zoom in');
assert.equal(f.c.pendingUserTextResponse, true);
await f.start('C');
assert.equal(f.c.pendingUserTextResponse, false);
await f.done('B-response');
assert.equal(f.sent.filter((x) => x.type === 'response.create').length, 0);
} finally {
f.close();
}
});
for (const id of [undefined, null, '', 42, {}, []]) {
test(`OpenAI missing or invalid provider identity is inert - ${JSON.stringify(id)}`, async () => {
const f = fixture({ pending: true });
let action;
try {
action = f.live();
await tick();
const before = f.snapshot();
const token = f.calls[0].options.userTurn;
await f.start(id);
assert.deepEqual(f.snapshot(), before);
assert.equal(readVoiceUserTurnText(token), 'zoom in a little');
assert.equal(f.calls[0].options.signal.aborted, false);
f.release();
await action;
} finally {
f.release();
if (action) await action;
f.close();
}
});
}
for (const kind of [
'currentTarget',
'target',
'closed',
'ending',
'progress-id',
'progress-tag',
]) {
test(`OpenAI non-conversation onset is inert and does not consume its identity - ${kind}`, async () => {
const f = fixture();
try {
await f.start('B');
if (kind === 'closed') f.c.dc.readyState = 'closed';
if (kind === 'ending') f.c.costCapStopped = true;
if (kind === 'progress-id')
f.c._turns.progressResponseIds.add('progress');
const before = f.snapshot();
const payload = {
type: 'input_audio_buffer.speech_started',
item_id: 'C',
...(kind === 'progress-id' ? { response_id: 'progress' } : {}),
...(kind === 'progress-tag'
? { response: { metadata: { gev: 'progress' } } }
: {}),
};
if (kind === 'target')
await f.c.handleRealtimeEvent({
data: JSON.stringify(payload),
target: f.channel(),
});
else
await f.event(payload, kind === 'currentTarget' ? f.channel() : f.c.dc);
assert.deepEqual(f.snapshot(), before);
assert.equal(f.c._turns.admittedSpeechItemIds.has('C'), false);
f.c.dc.readyState = 'open';
f.c.costCapStopped = false;
await f.start('C');
assert.equal(f.c._turns.admittedSpeechItemIds.has('C'), true);
} finally {
f.close();
}
});
}
test('OpenAI supports event fixtures without source metadata while production listeners enforce ownership', async () => {
const f = fixture();
try {
await f.c.handleRealtimeEvent({
data: JSON.stringify({
type: 'input_audio_buffer.speech_started',
item_id: 'B',
}),
});
await f.transcript('B', 'zoom in');
await f.commit('B');
await f.created('B-response');
assert.equal(
readVoiceUserTurnText(
f.c._turns.userTurnIntent.forResponse('B-response'),
),
'zoom in',
);
} finally {
f.close();
}
});
test('OpenAI Stop/new lifetime accepts the same ID once and excludes the obsolete channel', async () => {
const f = fixture();
try {
await f.start('B');
const oldChannel = f.c.dc;
f.c.stop();
assert.equal(f.c._turns.admittedSpeechItemIds.size, 0);
f.c.dc = f.channel();
f.c.status = 'listening';
const before = f.snapshot();
await f.event(
{ type: 'input_audio_buffer.speech_started', item_id: 'B' },
oldChannel,
);
assert.deepEqual(f.snapshot(), before);
await f.start('B');
await f.transcript('B', 'new lifetime');
await f.commit('B');
await f.created('fresh-B');
const token = f.c._turns.userTurnIntent.forResponse('fresh-B');
await f.start('B');
assert.equal(readVoiceUserTurnText(token), 'new lifetime');
} finally {
f.close();
}
});
for (const boundary of ['user-turn-start', 'interruption']) {
test(`OpenAI Stop from ${boundary} observer cannot continue stale admission`, async () => {
let armed = false;
const f = fixture({
onEvent: (event, c) => {
if (armed && event.type === boundary) c.stop();
},
});
try {
armed = true;
await f.start('B');
assert.equal(f.c.dc, null);
assert.equal(f.c.userTurnPending, false);
assert.equal(f.c.pendingUserTextResponse, false);
assert.equal(f.c._turns.admittedSpeechItemIds.size, 0);
assert.equal(f.snapshot().effects.radio, 0);
assert.equal(f.snapshot().effects.pointer, 0);
assert.equal(f.snapshot().effects.metrics, 0);
const expected = boundary === 'user-turn-start' ? 1 : 2;
assert.equal(
f.events.filter((e) =>
['user-turn-start', 'interruption'].includes(e.type),
).length,
expected,
);
} finally {
f.close();
}
});
}
test('OpenAI a token observer Stop prevents tracker resurrection by the same onset', async () => {
const f = fixture();
try {
await f.start('B');
await f.transcript('B', 'zoom in');
await f.commit('B');
await f.created('B-response');
const token = f.c._turns.userTurnIntent.forResponse('B-response');
subscribeVoiceUserTurn(token, (value) => {
if (!readVoiceUserTurnText(value)) f.c.stop();
});
await f.start('C');
assert.equal(f.c.dc, null);
assert.equal(
f.c._turns.userTurnIntent.transcript('C', 'late input'),
false,
);
assert.equal(f.c.userTurnPending, false);
assert.equal(f.c._turns.admittedSpeechItemIds.size, 0);
} finally {
f.close();
}
});
test('OpenAI opaque provider IDs are compared exactly without guessed ordering or arbitrary length rules', async () => {
const f = fixture();
try {
for (const id of [
'z-last-looking',
'a-first-looking',
'opaque-' + 'x'.repeat(4096),
]) {
await f.start(id);
await f.transcript(id, id);
await f.commit(id);
await f.created('reply-' + id.length);
assert.equal(
readVoiceUserTurnText(
f.c._turns.userTurnIntent.forResponse('reply-' + id.length),
),
id,
);
await f.done('reply-' + id.length);
}
} finally {
f.close();
}
});
test('OpenAI late original commits keep their response slots; rejecting a repeated start does not reorder them', async () => {
const f = fixture();
try {
await f.start('A');
await f.start('B');
await f.start('A');
await f.commit('A');
await f.transcript('B', 'current B');
await f.commit('B');
await f.created('old-A');
assert.equal(f.c._turns.isSupersededResponse('old-A'), true);
await f.created('current-B');
assert.equal(
readVoiceUserTurnText(f.c._turns.userTurnIntent.forResponse('current-B')),
'current B',
);
} finally {
f.close();
}
});
for (const via of ['boundary', 'token']) {
test(`OpenAI a newer typed request from a ${via} observer owns the next real response`, async () => {
let armed = false;
const f = fixture({
onEvent: (event, c) => {
if (armed && via === 'boundary' && event.type === 'user-turn-start') {
armed = false;
c.sendTextCommand('newer typed request');
}
},
});
try {
await f.start('B');
await f.transcript('B', 'old B');
await f.commit('B');
await f.created('R0');
if (via === 'token')
subscribeVoiceUserTurn(
f.c._turns.userTurnIntent.forResponse('R0'),
(token) => {
if (armed && !readVoiceUserTurnText(token)) {
armed = false;
f.c.sendTextCommand('newer typed request');
}
},
);
armed = true;
await f.start('C');
await f.transcript('C', 'older C request');
await f.done('R0');
await f.clientCreated('typed-response');
await f.tool('typed-response', 'newest');
assert.equal(f.calls.length, 1);
assert.equal(
readVoiceUserTurnText(f.calls[0].options.userTurn),
'newer typed request',
);
assert.equal(
readVoiceUserTurnProvenance(f.calls[0].options.userTurn).inputKind,
'typed',
);
} finally {
f.close();
}
});
}
test('OpenAI reentrant replacement during onset cannot mutate the replacement lifetime', async () => {
let armed = false;
let f;
f = fixture({
onEvent: (event, c) => {
if (armed && event.type === 'user-turn-start') {
armed = false;
c.stop();
c.dc = f.channel();
c.status = 'listening';
}
},
});
try {
const oldChannel = f.c.dc;
armed = true;
await f.start('B');
assert.notEqual(f.c.dc, oldChannel);
assert.equal(f.c.userTurnPending, false);
assert.equal(f.c._turns.admittedSpeechItemIds.size, 0);
assert.equal(f.events.filter((e) => e.type === 'interruption').length, 0);
await f.start('B');
await f.transcript('B', 'replacement request');
await f.commit('B');
await f.created('replacement');
assert.equal(
readVoiceUserTurnText(
f.c._turns.userTurnIntent.forResponse('replacement'),
),
'replacement request',
);
} finally {
f.close();
}
});
test('OpenAI local push-to-talk interruption does not forget already admitted provider IDs', async () => {
const f = fixture();
try {
await f.start('B');
await f.transcript('B', 'old speech');
await f.commit('B');
await f.created('B-response');
assert.equal(f.c._turns.bargeIn(), true);
const before = f.snapshot();
await f.start('B');
assert.deepEqual(f.snapshot(), before);
await f.start('C');
await f.transcript('C', 'new speech');
await f.commit('C');
await f.created('C-response');
assert.equal(
readVoiceUserTurnText(
f.c._turns.userTurnIntent.forResponse('C-response'),
),
'new speech',
);
} finally {
f.close();
}
});
for (const via of ['boundary', 'token']) {
test(`OpenAI late committed onset superseded by a ${via} typed observer keeps stale and typed FIFO slots distinct`, async () => {
let armed = false;
const f = fixture({
onEvent: (event, c) => {
if (armed && via === 'boundary' && event.type === 'user-turn-start') {
armed = false;
c.sendTextCommand('newer typed request');
}
},
});
try {
await f.start('B');
await f.transcript('B', 'old B');
await f.commit('B');
await f.created('R0');
if (via === 'token')
subscribeVoiceUserTurn(
f.c._turns.userTurnIntent.forResponse('R0'),
(token) => {
if (armed && !readVoiceUserTurnText(token)) {
armed = false;
f.c.sendTextCommand('newer typed request');
}
},
);
armed = true;
await f.start('C');
assert.deepEqual(f.c._turns.userTurnIntent.reservationState(), {
pending: 0,
overflowed: false,
});
assert.equal(f.c.pendingUserTextResponse, true);
await f.transcript('C', 'older C request');
await f.commit('C');
assert.equal(f.c._turns.userTurnIntent.reservationState().pending, 1);
await f.done('R0');
assert.equal(f.c._turns.userTurnIntent.reservationState().pending, 2);
const beforeCreates = f.sent.filter(
(e) => e.type === 'response.create',
).length;
await f.created('C-response');
assert.equal(f.c._turns.isSupersededResponse('C-response'), true);
assert.equal(
f.sent.some(
(e) => e.type === 'response.cancel' && e.response_id === 'C-response',
),
true,
);
await f.tool('C-response', 'stale-C');
assert.equal(f.output().at(-1).superseded, true);
assert.equal(f.calls.length, 0);
assert.equal(f.c.pendingResponseInstructions, null);
assert.equal(
f.sent.filter((e) => e.type === 'response.create').length,
beforeCreates,
);
await f.done('C-response');
await f.clientCreated('typed-response');
const typed = f.c._turns.userTurnIntent.forResponse('typed-response');
assert.equal(readVoiceUserTurnText(typed), 'newer typed request');
await f.tool('typed-response', 'typed-T');
assert.equal(f.calls.length, 1);
assert.equal(f.calls[0].options.userTurn, typed);
assert.equal(readVoiceUserTurnText(typed), 'newer typed request');
await f.event({
type: 'response.done',
response: {
id: 'typed-response',
status: 'completed',
output: [{ type: 'function_call' }],
},
});
await f.clientCreated('typed-followup');
assert.equal(
f.c._turns.userTurnIntent.forResponse('typed-followup'),
typed,
);
await f.done('typed-followup');
await f.start('D');
await f.transcript('D', 'genuine next speech');
await f.commit('D');
await f.start('C');
await f.created('D-response');
await f.tool('D-response', 'genuine-D');
assert.equal(f.calls.length, 2);
assert.equal(
readVoiceUserTurnText(f.calls[1].options.userTurn),
'genuine next speech',
);
assert.equal(f.c._turns.userTurnIntent.reservationState().pending, 0);
} finally {
f.close();
}
});
}
+16 -4
View File
@@ -105,7 +105,10 @@ test('push-to-talk: the held gesture sends the target once, with the chip after
{ pushToTalk: true },
);
controller.beginPointerTurn('keydown');
await event({ type: 'input_audio_buffer.speech_started' });
await event({
type: 'input_audio_buffer.speech_started',
item_id: 'ptt-held-pointer',
});
assert.deepEqual(
pointer.calls.map(([at]) => at),
['keydown', 'speech_start'],
@@ -145,7 +148,10 @@ test('push-to-talk: a speech start after the release keeps the gesture referent'
});
controller.beginPointerTurn('keydown');
pointer.releaseHold();
await event({ type: 'input_audio_buffer.speech_started' });
await event({
type: 'input_audio_buffer.speech_started',
item_id: 'ptt-released-pointer',
});
assert.equal(
pointer.calls.length,
1,
@@ -158,7 +164,10 @@ test('push-to-talk: a speech start after the release keeps the gesture referent'
test('open mic: only a recent pointer counts, and the model learns only that one exists', async () => {
const pointer = fakePointer([aircraftSnapshot]);
const { events, event, items } = controllerFixture(pointer);
await event({ type: 'input_audio_buffer.speech_started' });
await event({
type: 'input_audio_buffer.speech_started',
item_id: 'open-mic-recent-pointer',
});
assert.equal(
pointer.calls[0][1],
POINTER_OPEN_MIC_RECENT_MS,
@@ -179,7 +188,10 @@ test('open mic: only a recent pointer counts, and the model learns only that one
test('open mic: the chip and reticle appear when a tool resolves the pointer, once', async () => {
const pointer = fakePointer([aircraftSnapshot]);
const { controller, events, event } = controllerFixture(pointer);
await event({ type: 'input_audio_buffer.speech_started' });
await event({
type: 'input_audio_buffer.speech_started',
item_id: 'open-mic-resolved-pointer',
});
assert.equal(controller.announcePointer(), true);
assert.equal(controller.announcePointer(), false);
assert.equal(
+21 -3
View File
@@ -424,7 +424,14 @@ function fixture(t, initialProvider) {
} else {
await receiveOpenAI({
type: 'response.created',
response: { id: 'radio-request' },
response: {
id: 'radio-request',
metadata: peers
.at(-1)
.channel.sent.findLast(
(event) => event.type === 'response.create',
).response.metadata,
},
});
await receiveOpenAI({
type: 'response.function_call_arguments.done',
@@ -435,11 +442,22 @@ function fixture(t, initialProvider) {
});
await receiveOpenAI({
type: 'response.done',
response: { id: 'radio-request', status: 'completed' },
response: {
id: 'radio-request',
status: 'completed',
output: [{ type: 'function_call' }],
},
});
await receiveOpenAI({
type: 'response.created',
response: { id: 'radio-confirmation' },
response: {
id: 'radio-confirmation',
metadata: peers
.at(-1)
.channel.sent.findLast(
(event) => event.type === 'response.create',
).response.metadata,
},
});
await receiveOpenAI({
type: 'response.done',
+10 -3
View File
@@ -319,19 +319,26 @@ export class RealtimeConnection {
}
}
sendRealtimeEvent(message, logEventName = 'client.event') {
if (!this.dc || this.dc.readyState !== 'open') return false;
sendRealtimeEvent(
message,
logEventName = 'client.event',
{ isCurrent = () => true } = {},
) {
const channel = this.dc;
if (!channel || channel.readyState !== 'open' || !isCurrent()) return false;
this.debugLog(logEventName, {
type: message?.type || null,
message,
});
if (this.dc !== channel || channel.readyState !== 'open' || !isCurrent())
return false;
// A dc.send() that exceeds the SCTP send-buffer / max message size throws.
// If that throw escaped it would abort handleRealtimeEvent BEFORE
// queueResponseCreate + setStatus('listening'), stranding the turn at
// EXECUTING. Swallow it and signal failure so callers can fall through
// without the offending payload (M13).
try {
this.dc.send(JSON.stringify(message));
channel.send(JSON.stringify(message));
return true;
} catch (error) {
this.debugLog('client.send.failed', {
+5 -3
View File
@@ -281,8 +281,8 @@ export class GevRealtimeController extends RealtimeFacade {
* repeated.
* @param {'keydown'|'speech_start'|'text'} at
*/
beginPointerTurn(at) {
if (!this.pointer) return null;
beginPointerTurn(at, { isCurrent = () => true } = {}) {
if (!this.pointer || !isCurrent()) return null;
this._viewport.beginTurn();
// A speech start after the Space release belongs to the gesture that
// just ended; it must not re-read a pointer that has since moved.
@@ -303,6 +303,7 @@ export class GevRealtimeController extends RealtimeFacade {
} catch {
snapshot = null;
}
if (!isCurrent()) return null;
const chip = pointerChip(snapshot);
this.pointerAnnounced = gesture && chip ? snapshot : null;
this.emitSessionEvent({
@@ -310,6 +311,7 @@ export class GevRealtimeController extends RealtimeFacade {
pointer:
gesture && chip ? { ...chip, screenPx: snapshot.screenPx } : null,
});
if (!isCurrent()) return null;
const item = !chip
? this.pointerSentKey && this.pointerSentKey !== POINTER_NONE_KEY
? { type: 'pointer_context', at, target: 'none' }
@@ -326,7 +328,7 @@ export class GevRealtimeController extends RealtimeFacade {
const { at: _at, ...content } = item;
const key = chip ? JSON.stringify(content) : POINTER_NONE_KEY;
if (key === this.pointerSentKey) return snapshot;
if (this._turns.notifyMapEvent(item)) {
if (this._turns.notifyMapEvent(item, { isCurrent }) && isCurrent()) {
this.pointerSentKey = key;
this.debugLog('pointer.context', {
at,
+34 -9
View File
@@ -75,6 +75,18 @@ function controllerFixture({
}
const types = (sent) => sent.map((message) => message.type);
const clientCreated = (f, id) =>
f.event({
type: 'response.created',
response: {
id,
metadata: f.sent.findLast(
(e) =>
e.type === 'response.create' &&
e.response?.metadata?.gev !== 'progress',
)?.response?.metadata,
},
});
function speechFallbackFixture() {
const calls = [];
@@ -401,7 +413,10 @@ test('user speech stops an audible progress line: cancel and clear its audio', a
await progressLine('resp_p');
await event({ type: 'output_audio_buffer.started', response_id: 'resp_p' });
sent.length = 0;
await event({ type: 'input_audio_buffer.speech_started' });
await event({
type: 'input_audio_buffer.speech_started',
item_id: 'speech-during-progress',
});
assert.deepEqual(types(sent), [
'response.cancel',
'output_audio_buffer.clear',
@@ -529,7 +544,7 @@ test('caption transcription is metered with the served transcription model', asy
test('tools report start, progress and finish to narration; Radio stays silent', async () => {
const calls = [];
const { controller, event } = controllerFixture({
const { controller, event, sent } = controllerFixture({
runner: async (name, args, options) => {
options.progress?.({ step: 'resolve', label: 'Zilker Park' });
return { ok: true, action: name };
@@ -549,7 +564,14 @@ test('tools report start, progress and finish to narration; Radio stays silent',
controller.sendTextCommand(
'Mark Zilker Park and set Radio volume to 20 percent',
);
await event({ type: 'response.created', response: { id: 'resp_t' } });
await event({
type: 'response.created',
response: {
id: 'resp_t',
metadata: sent.findLast((e) => e.type === 'response.create')?.response
?.metadata,
},
});
await event({
type: 'response.output_audio_transcript.delta',
response_id: 'resp_t',
@@ -599,7 +621,7 @@ test('OpenAI dispatch binds first-party input across a tool follow-up and retire
},
});
f.controller.sendTextCommand('What am I looking at?');
await f.event({ type: 'response.created', response: { id: 'tools' } });
await clientCreated(f, 'tools');
await f.event({
type: 'response.function_call_arguments.done',
response_id: 'tools',
@@ -618,7 +640,7 @@ test('OpenAI dispatch binds first-party input across a tool follow-up and retire
assert.equal(calls[0].text, 'What am I looking at?');
assert.equal(calls[0].options.responseId, 'tools');
const token = calls[0].options.userTurn;
await f.event({ type: 'response.created', response: { id: 'ack' } });
await clientCreated(f, 'ack');
assert.equal(
observed.find(
(event) => event.type === 'response-start' && event.responseId === 'ack',
@@ -797,7 +819,7 @@ test('OpenAI cancels a stale response reservation and keeps its successor curren
});
f.controller.sendTextCommand('old turn');
f.controller.sendTextCommand('fresh turn');
await f.event({ type: 'response.created', response: { id: 'stale' } });
await clientCreated(f, 'stale');
assert.equal(
f.sent.findLast((message) => message.type === 'response.cancel')
?.response_id,
@@ -815,7 +837,7 @@ test('OpenAI cancels a stale response reservation and keeps its successor curren
type: 'response.done',
response: { id: 'stale', status: 'cancelled' },
});
await f.event({ type: 'response.created', response: { id: 'fresh' } });
await clientCreated(f, 'fresh');
await f.event({
type: 'response.function_call_arguments.done',
response_id: 'fresh',
@@ -838,7 +860,7 @@ test('OpenAI cancellation never publishes a stale resolved tool payload', async
},
});
f.controller.sendTextCommand('What is in the current view?');
await f.event({ type: 'response.created', response: { id: 'old' } });
await clientCreated(f, 'old');
const pending = f.event({
type: 'response.function_call_arguments.done',
response_id: 'old',
@@ -900,7 +922,7 @@ for (const [label, cancel] of [
},
});
f.controller.sendTextCommand('What is in the current view?');
await f.event({ type: 'response.created', response: { id: 'old' } });
await clientCreated(f, 'old');
const pending = f.event({
type: 'response.function_call_arguments.done',
response_id: 'old',
@@ -1157,6 +1179,9 @@ test('an untracked tool refusal cannot erase restart-required state when its fol
f.controller._turns.userTurnIntent.reservationState().pending,
16,
);
// No current input: an untracked refusal may speak neutral guidance, but
// must not borrow a different live typed request just to fill the queue.
f.controller._turns.userTurnIntent.invalidate();
await f.event({
type: 'response.function_call_arguments.done',
call_id: 'untracked-overflow',
+6 -1
View File
@@ -674,7 +674,12 @@ test('late action or viewport completion cannot resume a stopped or replacement
await controller.handleRealtimeEvent({
data: JSON.stringify({
type: 'response.created',
response: { id: 'owned-response' },
response: {
id: 'owned-response',
metadata: controller.dc.sent.findLast(
(e) => e.type === 'response.create',
)?.response?.metadata,
},
}),
});
const pending = controller.handleRealtimeEvent({
+11 -11
View File
@@ -365,23 +365,23 @@ export class RealtimeRadio {
epoch: this.radioHandoffEpoch,
responseId: responseId || null,
};
// Release only the old handoff before abort listeners can create a new one.
const shouldStopPlayback = this.radioHandoffInFlight;
const attemptId = this.radioHandoffAttemptId;
const radioLayer = this.radioLayer;
this.pendingRadioPlaybackResult = null;
this.radioHandoffInFlightResult = null;
this.radioToolHandoffReservations.clear();
this.radioHandoffDeferredByReservation = false;
this.radioHandoffInFlight = false;
this.radioHandoffAttemptId = null;
if (abortTools) {
this.abortTools();
} else if (abortRadioSiblings) {
this.abortRadioSiblingTools();
}
this.pendingRadioPlaybackResult = null;
this.radioHandoffInFlightResult = null;
this.radioToolHandoffReservations.clear();
this.radioHandoffDeferredByReservation = false;
const shouldStopPlayback = this.radioHandoffInFlight;
const attemptId = this.radioHandoffAttemptId;
// Release ownership before Stop synchronously notifies playback observers;
// the resulting callback is then idempotent instead of re-entering Stop.
this.radioHandoffInFlight = false;
this.radioHandoffAttemptId = null;
if (shouldStopPlayback) {
this.radioLayer?.stopPlayback?.({ origin: 'voice-cleanup', attemptId });
radioLayer?.stopPlayback?.({ origin: 'voice-cleanup', attemptId });
}
}
+236 -39
View File
@@ -72,7 +72,9 @@ export class RealtimeTurns {
this.responseCreatePending = false;
this.userTurnPending = false;
this.pendingResponseInstructions = null;
this.pendingResponseRequest = null;
this.pendingUserTextResponse = false;
this.pendingUserTextRequest = null;
this.activeResponseId = null;
this.supersededResponseIds = new Set();
this.activeToolAbortControllers = new Set();
@@ -87,7 +89,15 @@ export class RealtimeTurns {
this.progressResponseIds = new Set();
this.bestEffortEventIds = new Set();
this.bestEffortSequence = 0;
this.userTurnIntent = createVoiceUserTurnTracker();
this.userTurnIntent = createVoiceUserTurnTracker({ keyedRequests: true });
// Retain exact provider IDs for this connection lifetime: evicting older
// IDs would let delayed starts revoke a newer input again. Stop/reset
// releases the ledger; it is not a rolling-window or presentation cache.
this.admittedSpeechItemIds = new Set();
// Semantic response admission survives completed binding-map eviction.
// Exact IDs are released at the actual owned connection lifetime end.
this.admittedResponseIds = new Set();
this.inputGeneration = 0;
this.speechTranscriptionEnabled = true;
this.responseRequestSequence = 0;
this.outputClearReasons = [];
@@ -120,7 +130,7 @@ export class RealtimeTurns {
* payload is serialized JSON, so place names stay structured DATA (the same
* injection hygiene as failedLabels), never instruction-bearing prose.
*/
notifyMapEvent(payload) {
notifyMapEvent(payload, options) {
if (!this.dc || this.dc.readyState !== 'open') return false;
return this.sendRealtimeEvent(
{
@@ -132,6 +142,7 @@ export class RealtimeTurns {
},
},
'client.map_event',
options,
);
}
@@ -141,26 +152,41 @@ export class RealtimeTurns {
}
const cleanText = String(text || '').trim();
if (!cleanText) return;
const generation = ++this.inputGeneration;
const channel = this.dc;
const owned = () =>
this.inputGeneration === generation &&
this.ownsConversation(channel) &&
!this.isSessionEnding();
this.cancelRadioHandoff({ abortTools: true });
if (!owned()) return false;
const responseId = this.activeResponseId;
const audioPlaying = this.assistantAudioPlaying;
const interruptsForProviderGuidance = Boolean(
readVoiceProviderRequestText(cleanText),
);
this.supersedeActiveResponseForUserTurn();
if (interruptsForProviderGuidance && this.responseActive)
this.cancelResponse(responseId);
if (interruptsForProviderGuidance && audioPlaying)
this.clearOutputBuffer('provider-guidance');
if (interruptsForProviderGuidance && this.responseActive) {
this.cancelResponse(responseId, { isCurrent: owned });
if (!owned()) return false;
}
if (interruptsForProviderGuidance && audioPlaying) {
this.clearOutputBuffer('provider-guidance', { isCurrent: owned });
if (!owned()) return false;
}
this.emitSessionEvent({ type: 'user-turn-start' });
if (!owned()) return false;
this.userTurnIntent.text(cleanText);
this.metrics?.userText();
// New intent: stop any progress line, then start the new turn's timing.
if (!owned()) return false;
const request = this.userTurnIntent.captureRequest();
this.metrics?.userText(owned);
if (!owned()) return false;
this.narration?.cancel();
if (!owned()) return false;
this.narration?.userTurnEnded();
// A typed turn gets the same pointer context as a spoken one; it lands
// before the text so the model reads "this" with its referent.
this.beginPointerTurn?.('text');
if (!owned()) return false;
this.beginPointerTurn?.('text', { isCurrent: owned });
if (!owned()) return false;
const itemEvent = {
type: 'conversation.item.create',
item: {
@@ -169,11 +195,16 @@ export class RealtimeTurns {
content: [{ type: 'input_text', text: cleanText }],
},
};
if (!this.sendRealtimeEvent(itemEvent, 'client.user_text')) {
this.userTurnIntent.invalidate();
if (
!this.sendRealtimeEvent(itemEvent, 'client.user_text', {
isCurrent: owned,
})
) {
if (owned() && request.isCurrent()) this.userTurnIntent.invalidate();
return false;
}
this.requestUserTextResponse();
if (!owned() || !request.isCurrent()) return false;
this.requestUserTextResponse(request);
}
/**
@@ -202,6 +233,7 @@ export class RealtimeTurns {
}
}
this.pendingResponseInstructions = null;
this.pendingResponseRequest = null;
}
/**
@@ -226,13 +258,22 @@ export class RealtimeTurns {
* active response finishes.
* @returns {void}
*/
requestUserTextResponse() {
requestUserTextResponse(
request = this.pendingUserTextRequest ||
this.userTurnIntent.captureRequest(),
) {
if (!request.isCurrent()) {
this.pendingUserTextResponse = false;
this.pendingUserTextRequest = null;
return;
}
if (
this.responseActive ||
this.responseCreatePending ||
this.progressGenerating()
) {
this.pendingUserTextResponse = true;
this.pendingUserTextRequest = request;
this.debugLog('response.create.deferred_user_text', {
responseActive: this.responseActive,
responseCreatePending: this.responseCreatePending,
@@ -241,14 +282,20 @@ export class RealtimeTurns {
}
if (!this.dc || this.dc.readyState !== 'open') return;
this.pendingUserTextResponse = false;
this.pendingUserTextRequest = null;
this.responseCreatePending = true;
const eventId = `gev_response_${++this.responseRequestSequence}`;
const channel = this.dc;
const sent = this.sendRealtimeEvent(
{ type: 'response.create', event_id: eventId },
{
type: 'response.create',
event_id: eventId,
response: { metadata: { gev_request_id: eventId } },
},
'client.response_create.user_text',
);
if (sent) {
this.userTurnIntent.expectResponse(eventId);
if (sent && this.ownsConversation(channel) && !this.isSessionEnding()) {
request.reserve(eventId);
this.endSessionForReservationOverflow();
} else this.responseCreatePending = false;
}
@@ -269,6 +316,42 @@ export class RealtimeTurns {
return this.dc === channel && channel?.readyState === 'open';
}
rejectSpeechStart(event, payload, channel) {
const source = event.currentTarget ?? event.target;
if (source != null && source !== channel) return 'channel';
if (!this.ownsConversation(channel) || this.isSessionEnding())
return 'session';
if (this.isProgressEvent(payload)) return 'progress';
const id = payload.item_id;
if (typeof id !== 'string' || !id) return 'item-id';
if (this.admittedSpeechItemIds.has(id)) return 'duplicate';
this.admittedSpeechItemIds.add(id);
return null;
}
rejectResponseCreated(event, payload, channel) {
const source = event.currentTarget ?? event.target;
if (source && source !== channel) return 'transport';
if (!this.ownsConversation(channel) || this.isSessionEnding())
return 'closed';
const id = payload.response?.id;
if (typeof id !== 'string' || !id) return 'invalid-response-id';
if (this.admittedResponseIds.has(id)) return 'duplicate';
this.admittedResponseIds.add(id);
return null;
}
rememberSupersededSpeechStart(payload, channel) {
// Never carry a retired onset across Stop/reset or connection replacement.
// This only remembers an input already admitted in the owned lifetime.
if (
this.ownsConversation(channel) &&
!this.isSessionEnding() &&
this.admittedSpeechItemIds.has(payload.item_id)
)
this.userTurnIntent.rememberSupersededSpeech(payload.item_id);
}
async handleRealtimeEvent(event) {
await this.processRealtimeEvent(event);
}
@@ -281,6 +364,37 @@ export class RealtimeTurns {
} catch {
return;
}
const speechStart = payload?.type === 'input_audio_buffer.speech_started';
const responseCreated = payload?.type === 'response.created';
const responseAdmission = this.admittedResponseIds;
const ownsResponseCreated = () =>
this.admittedResponseIds === responseAdmission &&
this.ownsConversation(eventChannel) &&
!this.isSessionEnding();
if (responseCreated) {
const rejection = this.rejectResponseCreated(
event,
payload,
eventChannel,
);
if (rejection) {
this.debugLog('voice.response_created.rejected', { reason: rejection });
return;
}
}
let inputGeneration = this.inputGeneration;
const ownsSpeechStart = () =>
this.inputGeneration === inputGeneration &&
this.ownsConversation(eventChannel) &&
!this.isSessionEnding();
if (speechStart) {
const rejection = this.rejectSpeechStart(event, payload, eventChannel);
if (rejection) {
this.debugLog('voice.speech_start.rejected', { reason: rejection });
return;
}
inputGeneration = ++this.inputGeneration;
}
const progressEvent = this.isProgressEvent(payload);
if (!progressEvent) {
if (payload.type === 'output_audio_buffer.cleared') {
@@ -306,10 +420,24 @@ export class RealtimeTurns {
session.input_audio_transcription !== null;
}
if (payload.type === 'input_audio_buffer.speech_started') {
// A newer microphone turn supersedes a typed response that has not
// been sent yet. Flushing it later would reserve this speech record
// instead of the typed input it originally answered.
this.pendingUserTextResponse = false;
this.pendingUserTextRequest = null;
this.emitSessionEvent({ type: 'user-turn-start' });
if (!ownsSpeechStart()) {
this.rememberSupersededSpeechStart(payload, eventChannel);
return;
}
this.userTurnIntent.speechStarted(payload.item_id, {
transcriptionEnabled: this.speechTranscriptionEnabled,
isCurrent: ownsSpeechStart,
});
if (!ownsSpeechStart()) {
this.rememberSupersededSpeechStart(payload, eventChannel);
return;
}
}
if (payload.type === 'conversation.item.input_audio_transcription.failed')
this.userTurnIntent.transcriptionUnavailable(payload.item_id, 'failed');
@@ -330,9 +458,17 @@ export class RealtimeTurns {
if (this.endSessionForReservationOverflow()) return;
}
if (payload.type === 'response.created' && !progressEvent) {
responseBinding = this.userTurnIntent.bindResponse(
payload.response?.id,
);
const metadata = payload.response?.metadata;
const requestId =
metadata && Object.hasOwn(metadata, 'gev_request_id')
? typeof metadata.gev_request_id === 'string'
? metadata.gev_request_id
: false
: null;
responseBinding =
payload.response?.conversation_id === null
? 'untracked'
: this.userTurnIntent.bindResponse(payload.response?.id, requestId);
this.emitSessionEvent({
type: 'response-start',
responseId: payload.response?.id,
@@ -340,10 +476,18 @@ export class RealtimeTurns {
});
}
}
if (speechStart && !ownsSpeechStart()) return;
if (responseCreated && !ownsResponseCreated()) return;
const sessionEvent = realtimeSessionEvent(payload);
if (sessionEvent) this.emitSessionEvent(sessionEvent);
if (speechStart && !ownsSpeechStart()) return;
if (responseCreated && !ownsResponseCreated()) return;
this.metrics?.serverEvent(payload);
if (speechStart && !ownsSpeechStart()) return;
if (responseCreated && !ownsResponseCreated()) return;
this.observeSpeech(payload);
if (speechStart && !ownsSpeechStart()) return;
if (responseCreated && !ownsResponseCreated()) return;
// Streaming deltas are skipped: their `.done` events carry the full text,
// and logging every token spent the debug sink's per-minute write budget,
// dropping later turns' records.
@@ -380,14 +524,20 @@ export class RealtimeTurns {
return;
}
if (payload.error?.code === 'conversation_already_has_active_response') {
this.userTurnIntent.rejectPendingResponse(echoedId);
this.cancelRadioHandoff({ abortTools: true });
this.responseActive = true;
this.responseCreatePending = false;
this.pendingResponseInstructions = null;
this.pendingResponseRequest = null;
// Never replay: the rejected turn is dropped, not retried. Re-arming
// here is how one collision becomes the same sentence twice.
this.pendingUserTextResponse = false;
this.pendingUserTextRequest = null;
// Cancellation and token retirement may synchronously submit new input.
// Finish old-request cleanup first so those successors are preserved.
this.cancelRadioHandoff({ abortTools: true });
this.userTurnIntent.rejectPendingResponse(echoedId);
if (!this.ownsConversation(eventChannel) || this.isSessionEnding())
return;
console.warn('[GEV Realtime] Skipped overlapping response.create');
this.debugLog('response.create.skipped_active', {
eventId: payload.event_id,
@@ -412,12 +562,16 @@ export class RealtimeTurns {
});
return;
}
this.userTurnIntent.rejectPendingResponse(echoedId);
this.responseActive = false;
this.responseCreatePending = false;
this.pendingResponseInstructions = null;
this.pendingResponseRequest = null;
this.pendingUserTextResponse = false;
this.pendingUserTextRequest = null;
this.cancelRadioHandoff({ abortTools: true });
this.userTurnIntent.rejectPendingResponse(echoedId);
if (!this.ownsConversation(eventChannel) || this.isSessionEnding())
return;
this.reportError('Realtime API', payload.error, {
eventId: payload.event_id,
type: payload.error?.type,
@@ -442,16 +596,22 @@ export class RealtimeTurns {
return;
}
if (payload.type === 'input_audio_buffer.speech_started') {
if (speechStart) {
if (!ownsSpeechStart()) return;
this.userTurnPending = true;
this.pendingResponseInstructions = null;
this.pendingResponseRequest = null;
this.cancelRadioHandoff({ abortTools: true });
if (!ownsSpeechStart()) return;
this.setVoiceSpeaker('user');
if (!ownsSpeechStart()) return;
// Point-and-ask: read the pointer as speech starts (open mic), or reuse
// the push-to-talk keydown snapshot, before the user audio commits.
this.beginPointerTurn?.('speech_start');
if (!ownsSpeechStart()) return;
}
this.updateResponseState(payload);
if (responseCreated && !ownsResponseCreated()) return;
if (payload.type === 'response.created' && responseBinding === 'stale') {
this.supersedeActiveResponseForUserTurn();
this.cancelResponse(this.activeResponseId);
@@ -535,6 +695,7 @@ export class RealtimeTurns {
return;
}
const responseRequest = this.userTurnIntent.captureRequest(toolResponseId);
const responseIntent =
await this.userTurnIntent.waitForResponse(toolResponseId);
if (!this.ownsConversation(eventChannel) || this.isSessionEnding()) return;
@@ -569,7 +730,10 @@ export class RealtimeTurns {
this.sendToolOutput(call.call_id || call.id, lastRefusal);
}
if (responseIntent.status !== 'stale' && lastRefusal)
this.queueResponseCreate(toolResultGuidance(lastRefusal));
this.queueResponseCreate(
toolResultGuidance(lastRefusal),
responseRequest,
);
if (!this.ownsConversation(eventChannel) || this.isSessionEnding())
return;
this.setStatus('listening', 'Ask or command');
@@ -874,6 +1038,7 @@ export class RealtimeTurns {
// result follows Radio in the same multi-intent response.
this.queueResponseCreate(
toolResultGuidance(this.radio.pendingRadioPlaybackResult || lastResult),
responseRequest,
);
if (!this.ownsConversation(eventChannel)) return;
}
@@ -1073,17 +1238,18 @@ export class RealtimeTurns {
}
}
clearOutputBuffer(reason = null) {
clearOutputBuffer(reason = null, options) {
const sent = this.sendBestEffort(
{ type: 'output_audio_buffer.clear' },
'client.output_audio_clear',
options,
);
if (sent) this.outputClearReasons.push(reason);
return sent;
}
/** Send a client event whose rejection is expected and harmless. */
sendBestEffort(message, logEventName) {
sendBestEffort(message, logEventName, options) {
const eventId = `gev_be_${++this.bestEffortSequence}`;
this.bestEffortEventIds.add(eventId);
while (this.bestEffortEventIds.size > 16)
@@ -1093,16 +1259,18 @@ export class RealtimeTurns {
return this.sendRealtimeEvent(
{ ...message, event_id: eventId },
logEventName,
options,
);
}
cancelResponse(responseId) {
cancelResponse(responseId, options) {
return this.sendBestEffort(
{
type: 'response.cancel',
...(responseId ? { response_id: responseId } : {}),
},
'client.response_cancel',
options,
);
}
@@ -1123,22 +1291,35 @@ export class RealtimeTurns {
Boolean(this.progressResponse);
if (!speaking) return false;
if (!this.radio.mayVoiceClaimSpeaker()) return false;
const generation = ++this.inputGeneration;
const channel = this.dc;
const owned = () =>
this.inputGeneration === generation &&
this.ownsConversation(channel) &&
!this.isSessionEnding();
this.userTurnIntent.invalidate();
if (!owned()) return false;
this.emitSessionEvent({ type: 'interruption', reason: 'push-to-talk' });
if (!owned()) return false;
const responseId = this.activeResponseId;
const audioPlaying = this.assistantAudioPlaying;
this.narration?.cancel();
if (!owned()) return false;
this.metrics?.bargeIn();
if (!owned()) return false;
this.pendingResponseInstructions = null;
this.pendingResponseRequest = null;
if (this.responseActive) {
this.supersedeActiveResponseForUserTurn();
this.cancelResponse(responseId);
this.cancelResponse(responseId, { isCurrent: owned });
if (!owned()) return false;
}
if (audioPlaying) {
this.clearOutputBuffer();
this.clearOutputBuffer(null, { isCurrent: owned });
if (!owned()) return false;
}
this.debugLog('voice.barge_in', { responseId, audioPlaying });
return true;
return owned();
}
sendToolOutput(callId, result) {
@@ -1227,7 +1408,11 @@ export class RealtimeTurns {
}
}
queueResponseCreate(instructions) {
queueResponseCreate(
instructions,
request = this.userTurnIntent.captureRequest(),
) {
if (!request.isCurrent()) return;
if (this.userTurnPending) {
this.debugLog('response.create.skipped_user_turn', {
instructions: instructions || null,
@@ -1236,6 +1421,7 @@ export class RealtimeTurns {
}
this.pendingResponseInstructions =
instructions || 'Briefly respond once. Do not repeat yourself.';
this.pendingResponseRequest = request;
if (!this.responseActive && !this.responseCreatePending)
this.flushPendingResponse();
}
@@ -1251,34 +1437,43 @@ export class RealtimeTurns {
this.dc.readyState !== 'open'
)
return;
const request = this.pendingResponseRequest;
const instructions = this.pendingResponseInstructions;
this.pendingResponseInstructions = null;
this.pendingResponseRequest = null;
if (!request?.isCurrent()) return;
this.responseCreatePending = true;
const eventId = `gev_response_${++this.responseRequestSequence}`;
const channel = this.dc;
const sent = this.sendRealtimeEvent(
{
type: 'response.create',
event_id: eventId,
response: { instructions },
response: { instructions, metadata: { gev_request_id: eventId } },
},
'client.response_create.tool_followup',
);
if (sent) {
this.userTurnIntent.expectResponse(eventId);
if (sent && this.ownsConversation(channel) && !this.isSessionEnding()) {
request.reserve(eventId);
this.endSessionForReservationOverflow();
} else this.responseCreatePending = false;
}
abortTools() {
for (const controller of this.activeToolAbortControllers)
controller.abort();
const controllers = [...this.activeToolAbortControllers];
const resources = this.turnResourceAbort;
this.activeToolAbortControllers.clear();
this.turnResourceAbort.abort();
this.turnResourceAbort = new AbortController();
this.radio.clearTools();
for (const controller of controllers) controller.abort();
resources.abort();
}
reset() {
this.inputGeneration++;
this.admittedSpeechItemIds.clear();
this.admittedResponseIds.clear();
this.admittedResponseIds = new Set();
this.userTurnIntent.reset();
this.speechTranscriptionEnabled = true;
this.processedCalls.clear();
@@ -1286,7 +1481,9 @@ export class RealtimeTurns {
this.responseCreatePending = false;
this.userTurnPending = false;
this.pendingResponseInstructions = null;
this.pendingResponseRequest = null;
this.pendingUserTextResponse = false;
this.pendingUserTextRequest = null;
this.activeResponseId = null;
this.supersededResponseIds.clear();
this.narration?.cancel();
+5 -1
View File
@@ -109,7 +109,11 @@ for (const malicious of [false, true]) {
controller.sendTextCommand('Zoom in');
await event({
type: 'response.created',
response: { id: 'zoom-response' },
response: {
id: 'zoom-response',
metadata: sent.findLast((event) => event.type === 'response.create')
.response.metadata,
},
});
await event({
type: 'response.function_call_arguments.done',
+2 -1
View File
@@ -41,8 +41,9 @@ export class TurnMetrics {
}
/** Begin a turn for typed input; the command is its end of speech. */
userText() {
userText(isCurrent = () => true) {
this.close('next_turn');
if (!isCurrent()) return;
this.open('text');
this.turn.eouAt = this.now();
}
File diff suppressed because it is too large. Load diff
+61 -9
View File
@@ -32,6 +32,7 @@ export function subscribeVoiceUserTurn(token, listener) {
export function createVoiceUserTurnTracker({
setTimer = setTimeout,
clearTimer = clearTimeout,
keyedRequests = false,
} = {}) {
let current = null;
const responses = new Map();
@@ -39,6 +40,7 @@ export function createVoiceUserTurnTracker({
const pending = [];
const waiters = new Set();
let overflow = false;
let resetGeneration = 0;
const retireCurrent = () => {
const previous = current;
current = null;
@@ -70,12 +72,15 @@ export function createVoiceUserTurnTracker({
}
};
const reset = () => {
resetGeneration++;
settleWaiters({ status: 'stale', userTurn: undefined });
retireCurrent();
const previous = current;
current = null;
responses.clear();
inputs.clear();
pending.length = 0;
overflow = false;
notifyVoiceUserTurn(previous, { retire: true });
};
const begin = (
itemId,
@@ -86,7 +91,7 @@ export function createVoiceUserTurnTracker({
// Keep pending response reservations for older turns. Their delayed
// response.created must consume their own slot, never the newer input.
settleWaiters({ status: 'stale', userTurn: undefined });
retireCurrent();
const previous = current;
responses.clear();
const token = Object.freeze({});
const record = {
@@ -104,6 +109,8 @@ export function createVoiceUserTurnTracker({
inputs.set(itemId, record);
while (inputs.size > 16) inputs.delete(inputs.keys().next().value);
}
notifyVoiceUserTurn(previous, { retire: true });
return current === record ? token : undefined;
};
const reserve = (record, requestId = null) => {
if (overflow) return;
@@ -121,8 +128,8 @@ export function createVoiceUserTurnTracker({
},
invalidate() {
settleWaiters({ status: 'stale', userTurn: undefined });
retireCurrent();
responses.clear();
retireCurrent();
},
rejectPendingResponse(requestId) {
if (!requestId) return false;
@@ -130,17 +137,26 @@ export function createVoiceUserTurnTracker({
if (index < 0) return false;
const [{ record }] = pending.splice(index, 1);
if (record === current) {
retireCurrent();
responses.clear();
retireCurrent();
}
return true;
},
text(text) {
begin(null, typeof text === 'string' ? text.trim() : '', 'typed');
return begin(null, typeof text === 'string' ? text.trim() : '', 'typed');
},
speechStarted(itemId, { transcriptionEnabled = true } = {}) {
// Provider adapters own duplicate/input admission before this transition.
speechStarted(
itemId,
{ transcriptionEnabled = true, isCurrent = () => true } = {},
) {
const generation = resetGeneration;
settleWaiters({ status: 'stale', userTurn: undefined });
retireCurrent();
// A synchronous token observer may Stop/replace the session. That reset
// owns the new lifetime. An observer can also submit newer typed input;
// this old onset must not overwrite either successor after notification.
if (resetGeneration !== generation || !isCurrent()) return;
responses.clear();
if (typeof itemId === 'string' && itemId)
begin(
@@ -195,6 +211,14 @@ export function createVoiceUserTurnTracker({
notifyVoiceUserTurn(current);
return true;
},
rememberSupersededSpeech(itemId) {
// An admitted onset can lose to newer local input during notification,
// before it becomes current. Its later real commit still needs a stale
// identity, but remembering it grants no token or response reservation.
if (typeof itemId !== 'string' || !itemId || inputs.has(itemId)) return;
inputs.set(itemId, { itemId, committed: false });
while (inputs.size > 16) inputs.delete(inputs.keys().next().value);
},
commitSpeech(itemId) {
// A provider input-buffer commit causes one automatic VAD response.
const record = inputs.get(itemId);
@@ -202,14 +226,42 @@ export function createVoiceUserTurnTracker({
if (record) record.committed = true;
reserve(record || null);
},
captureRequest(responseId = undefined) {
const record =
responseId === undefined ? current : responses.get(responseId) || null;
const generation = resetGeneration;
let reserved = false;
return Object.freeze({
isCurrent: () => generation === resetGeneration && record === current,
reserve(requestId) {
if (reserved || generation !== resetGeneration) return false;
reserved = true;
// Capture before transport callbacks, reserve only after send. A
// newer input makes this original record stale, never its successor.
reserve(record, requestId);
return true;
},
});
},
expectResponse(requestId = null) {
// Called only when the application sends a typed/tool-follow-up response
// request; generated out-of-band progress responses do not reserve one.
reserve(current, requestId);
},
bindResponse(id) {
bindResponse(id, requestId = null) {
if (!id || overflow) return null;
const record = pending.shift()?.record;
if (keyedRequests && responses.has(id))
return responses.get(id) === current ? 'current' : 'stale';
let index = 0;
if (keyedRequests) {
// Only server-returned client metadata can select a client request.
// Untagged automatic responses cannot borrow typed/follow-up slots.
if (requestId !== null && (typeof requestId !== 'string' || !requestId))
return 'stale';
index = pending.findIndex((slot) => slot.requestId === requestId);
if (index < 0) return requestId === null ? 'untracked' : 'stale';
}
const record = pending.splice(index, 1)[0]?.record;
if (!record) return 'untracked';
if (record !== current) return 'stale';
responses.set(id, record);
@@ -254,8 +306,8 @@ export function createVoiceUserTurnTracker({
// A function-call response can continue after its tool result. A final
// answer retires the request, so later automatic turns cannot reuse it.
if (record && record === current && !hasToolCalls) {
retireCurrent();
responses.clear();
retireCurrent();
}
},
reset,
+36
View File
@@ -459,3 +459,39 @@ test('reset invalidates tokens even when a later session reuses provider respons
assert.equal(readVoiceUserTurnText(old), undefined);
assert.equal(readVoiceUserTurnText(fresh), 'switch to OpenAI');
});
for (const transition of [
'text',
'invalidate',
'complete',
'reject',
'reset',
]) {
test(`retirement ${transition} cannot erase a response bound by its observer`, () => {
const t = createVoiceUserTurnTracker({ keyedRequests: true });
t.text('old');
t.expectResponse('old-key');
t.bindResponse('old-response', 'old-key');
const old = t.forResponse('old-response');
if (transition === 'reject') t.expectResponse('reject-key');
let next;
subscribeVoiceUserTurn(old, () => {
if (readVoiceUserTurnText(old) !== undefined) return;
t.text('newer observer');
t.expectResponse('new-key');
t.bindResponse('new-response', 'new-key');
next = t.forResponse('new-response');
});
if (transition === 'text') t.text('older outer');
if (transition === 'invalidate') t.invalidate();
if (transition === 'complete') t.completeResponse('old-response');
if (transition === 'reject') t.rejectPendingResponse('reject-key');
if (transition === 'reset') t.reset();
assert.equal(readVoiceUserTurnText(old), undefined);
assert.equal(
readVoiceUserTurnText(t.forResponse('new-response')),
'newer observer',
);
assert.equal(t.forResponse('new-response'), next);
});
}