Skip to content

fix(talkio): Harden provider interruption handling #1

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
abdufelsayed merged 1 commit into from
Apr 28, 2026
Merged

fix(talkio): Harden provider interruption handling #1

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
43 changes: 19 additions & 24 deletions packages/talkio/src/agent/actors/llm.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -38,6 +38,16 @@ export const llmActor = fromCallback<
const debug = config.debug ?? false;

let isAborted = false;
let isSettled = false;
let timeoutId: ReturnType<typeof setTimeout> | null = null;

const finishWithError = (error: Error) => {
if (isAborted || isSettled) return;
isSettled = true;
isAborted = true;
if (timeoutId) clearTimeout(timeoutId);
sendBack({ type: "_llm:error", error, timestamp: Date.now() });
};

const handleAbort = () => {
isAborted = true;
Expand All @@ -53,42 +63,33 @@ export const llmActor = fromCallback<
};
}

// Set up timeout
const timeoutMs = config.timeout?.llmMs;
let timeoutId: ReturnType<typeof setTimeout> | null = null;

if (timeoutMs && timeoutMs > 0) {
timeoutId = setTimeout(() => {
if (!isAborted) {
isAborted = true;
if (debug) console.error("[llm-actor] Timeout after", timeoutMs, "ms");
sendBack({
type: "_llm:error",
error: new Error(`LLM timeout after ${timeoutMs}ms`),
timestamp: Date.now(),
});
}
if (debug) console.error("[llm-actor] Timeout after", timeoutMs, "ms");
finishWithError(new Error(`LLM timeout after ${timeoutMs}ms`));
}, timeoutMs);
}

const ctx = {
messages,
token: (token: string) => {
if (isAborted) return;
if (isAborted || isSettled) return;
sendBack({ type: "_llm:token", token, timestamp: Date.now() });
},
sentence: (sentence: string, index: number) => {
if (isAborted) return;
if (isAborted || isSettled) return;
sendBack({ type: "_llm:sentence", sentence, index, timestamp: Date.now() });
},
complete: (fullText: string) => {
if (isAborted) return;
if (isAborted || isSettled) return;
isSettled = true;
if (timeoutId) clearTimeout(timeoutId);
sendBack({ type: "_llm:complete", fullText, timestamp: Date.now() });
},
error: (error: Error) => {
if (isAborted) return;
sendBack({ type: "_llm:error", error, timestamp: Date.now() });
finishWithError(error);
},
say: sayFn,
interrupt: interruptFn,
Expand All @@ -103,14 +104,8 @@ export const llmActor = fromCallback<
config.llm(ctx);
}
} catch (error) {
if (!isAborted) {
if (debug) console.error("[llm-actor] Error starting generation:", error);
sendBack({
type: "_llm:error",
error: error instanceof Error ? error : new Error(String(error)),
timestamp: Date.now(),
});
}
if (debug) console.error("[llm-actor] Error starting generation:", error);
finishWithError(error instanceof Error ? error : new Error(String(error)));
}

return () => {
Expand Down
7 changes: 7 additions & 0 deletions packages/talkio/src/agent/actors/stt.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -42,6 +42,13 @@ export const sttActor = fromCallback<

abortSignal.addEventListener("abort", handleAbort);

if (abortSignal.aborted) {
isAborted = true;
return () => {
abortSignal.removeEventListener("abort", handleAbort);
};
}

if (debug) console.log("[stt-actor] Starting STT provider with format:", inputFormat);

try {
Expand Down
48 changes: 22 additions & 26 deletions packages/talkio/src/agent/actors/tts.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,15 +27,26 @@ export const ttsActor = fromCallback<
{
config: NormalizedAgentConfig;
text: string;
requestId: string;
abortSignal: AbortSignal;
}
>(({ sendBack, input }) => {
const { config, text, abortSignal } = input;
const { config, text, requestId, abortSignal } = input;
const provider = config.tts;
const outputFormat = config.audio.output;
const debug = config.debug ?? false;

let isAborted = false;
let isSettled = false;
let timeoutId: ReturnType<typeof setTimeout> | null = null;

const finishWithError = (error: Error) => {
if (isAborted || isSettled) return;
isSettled = true;
isAborted = true;
if (timeoutId) clearTimeout(timeoutId);
sendBack({ type: "_tts:error", requestId, error, timestamp: Date.now() });
};

const handleAbort = () => {
isAborted = true;
Expand All @@ -51,51 +62,36 @@ export const ttsActor = fromCallback<
};
}

// Set up timeout
const timeoutMs = config.timeout?.ttsMs;
let timeoutId: ReturnType<typeof setTimeout> | null = null;

if (timeoutMs && timeoutMs > 0) {
timeoutId = setTimeout(() => {
if (!isAborted) {
isAborted = true;
if (debug) console.error("[tts-actor] Timeout after", timeoutMs, "ms");
sendBack({
type: "_tts:error",
error: new Error(`TTS timeout after ${timeoutMs}ms`),
timestamp: Date.now(),
});
}
if (debug) console.error("[tts-actor] Timeout after", timeoutMs, "ms");
finishWithError(new Error(`TTS timeout after ${timeoutMs}ms`));
}, timeoutMs);
}

try {
provider.synthesize(text, {
audioFormat: outputFormat,
audioChunk: (audio) => {
if (isAborted) return;
sendBack({ type: "_tts:chunk", audio, timestamp: Date.now() });
if (isAborted || isSettled) return;
sendBack({ type: "_tts:chunk", requestId, audio, timestamp: Date.now() });
},
complete: () => {
if (isAborted) return;
if (isAborted || isSettled) return;
isSettled = true;
if (timeoutId) clearTimeout(timeoutId);
sendBack({ type: "_tts:complete", timestamp: Date.now() });
sendBack({ type: "_tts:complete", requestId, timestamp: Date.now() });
},
error: (error) => {
if (isAborted) return;
sendBack({ type: "_tts:error", error, timestamp: Date.now() });
finishWithError(error);
},
signal: abortSignal,
});
} catch (error) {
if (!isAborted) {
if (debug) console.error("[tts-actor] Error starting synthesis:", error);
sendBack({
type: "_tts:error",
error: error instanceof Error ? error : new Error(String(error)),
timestamp: Date.now(),
});
}
if (debug) console.error("[tts-actor] Error starting synthesis:", error);
finishWithError(error instanceof Error ? error : new Error(String(error)));
}

return () => {
Expand Down
7 changes: 7 additions & 0 deletions packages/talkio/src/agent/actors/turn-detector.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -45,6 +45,13 @@ export const turnDetectorActor = fromCallback<

abortSignal.addEventListener("abort", handleAbort);

if (abortSignal.aborted) {
isAborted = true;
return () => {
abortSignal.removeEventListener("abort", handleAbort);
};
}

try {
provider.start({
turnEnd: (transcript) => {
Expand Down
20 changes: 18 additions & 2 deletions packages/talkio/src/agent/actors/vad.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -44,6 +44,13 @@ export const vadActor = fromCallback<

abortSignal.addEventListener("abort", handleAbort);

if (abortSignal.aborted) {
isAborted = true;
return () => {
abortSignal.removeEventListener("abort", handleAbort);
};
}

try {
provider.start({
speechStart: () => {
Expand All @@ -58,6 +65,10 @@ export const vadActor = fromCallback<
if (isAborted) return;
sendBack({ type: "_vad:probability", value, timestamp: Date.now() });
},
error: (error) => {
if (isAborted) return;
sendBack({ type: "_vad:error", error, timestamp: Date.now() });
},
signal: abortSignal,
});
} catch (error) {
Expand All @@ -77,8 +88,13 @@ export const vadActor = fromCallback<
try {
provider.processAudio(event.audio);
} catch (error) {
if (!isAborted && debug) {
console.error("[vad-actor] Error processing audio:", error);
if (!isAborted) {
if (debug) console.error("[vad-actor] Error processing audio:", error);
sendBack({
type: "_vad:error",
error: error instanceof Error ? error : new Error(String(error)),
timestamp: Date.now(),
});
}
}
}
Expand Down
2 changes: 2 additions & 0 deletions packages/talkio/src/agent/context.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -38,6 +38,7 @@ export interface AgentMachineContext {

llmRef: ActorRefFrom<typeof llmActor> | null;
ttsRef: ActorRefFrom<typeof ttsActor> | null;
activeTTSRequestId: string | null;
sentenceQueue: string[];
vadSource: "adapter" | "stt";
turnSource: "adapter" | "stt";
Expand Down Expand Up @@ -68,6 +69,7 @@ export function createInitialContext<
turnAbortController: null,
llmRef: null,
ttsRef: null,
activeTTSRequestId: null,
sentenceQueue: [],
vadSource: config.vad ? "adapter" : "stt",
turnSource: config.turnDetector ? "adapter" : "stt",
Expand Down
Loading
Loading