Heartbeat Thinking (arXiv:2610.26233): Speculative Reasoning & Asynchronous Tool Execution Saat Mendengarkan pada Realtime Voice Agents — Reduksi Latensi 36.4% dan Peluncuran Tool Pre-Speech End

// Production Implementation: Heartbeat Thinking Runtime Engine
// Implements Decoupled Periodic Thinker Invocations, Shared Execution State, and VAD Handoff

export interface ToolCallSpec {
  callId: string;
  toolName: string;
  arguments: Record;
  status: 'pending' | 'running' | 'completed' | 'cancelled';
  result?: unknown;
  dispatchedAt: number;
  completedAt?: number;
}

export interface ReasoningRecord {
  recordId: string;
  invocationIndex: number;
  partialTranscript: string;
  provisionalThought: string;
  timestamp: number;
}

export interface SharedExecutionState {
  completedReasoning: ReasoningRecord[];
  toolRegistry: Map;
  activeThinkerCount: number;
}

export class HeartbeatThinkingEngine {
  private state: SharedExecutionState = {
    completedReasoning: [],
    toolRegistry: new Map(),
    activeThinkerCount: 0,
  };

  private currentAbortController: AbortController | null = null;
  private heartbeatTimer: NodeJS.Timeout | null = null;
  private invocationCounter = 0;

  constructor(
    private readonly deltaIntervalMs = 500, // \Delta: Periode denyut Thinker
    private readonly maxConcurrencyK = 2,     // K: Batas paralelisme invokasi Thinker
    private readonly toolExecutor: (toolName: string, args: Record, signal: AbortSignal) => Promise
  ) {}

  public startListeningSession(): void {
    this.currentAbortController = new AbortController();
    this.state = {
      completedReasoning: [],
      toolRegistry: new Map(),
      activeThinkerCount: 0,
    };
    this.invocationCounter = 0;
  }

  // Dipanggil setiap kali interval \Delta terpicu selama streaming ASR berjalan
  public async onHeartbeatTick(currentPartialTranscript: string): Promise {
    if (!this.currentAbortController || this.currentAbortController.signal.aborted) return;

    // Concurrency throttle: jika K slot penuh, lewati tick tanpa memutus worker aktif
    if (this.state.activeThinkerCount >= this.maxConcurrencyK) {
      return;
    }

    const invocationId = ++this.invocationCounter;
    this.state.activeThinkerCount++;

    try {
      // Thinker membaca partial transcript terkini dan snapshot shared execution state
      const completedToolsSnapshot = Array.from(this.state.toolRegistry.values())
        .filter(t => t.status === 'completed')
        .map(t => ({ tool: t.toolName, args: t.arguments, result: t.result }));

      // Simulasi analisis spekulatif Thinker
      const speculativePlan = await this.evaluateThinkerStep(
        invocationId,
        currentPartialTranscript,
        completedToolsSnapshot,
        this.currentAbortController.signal
      );

      if (speculativePlan.reasoning) {
        this.state.completedReasoning.push({
          recordId: `rr-${invocationId}`,
          invocationIndex: invocationId,
          partialTranscript: currentPartialTranscript,
          provisionalThought: speculativePlan.reasoning,
          timestamp: Date.now(),
        });
      }

      // Dispatch tools secara asinkron tanpa memblokir thread spekulasi berikutnya
      for (const call of speculativePlan.requestedTools) {
        const deduplicationKey = `${call.toolName}:${JSON.stringify(call.arguments)}`;
        if (this.state.toolRegistry.has(deduplicationKey)) {
          continue; // Hindari pemanggilan ganda identik
        }

        const toolSpec: ToolCallSpec = {
          callId: `tool-${Date.now()}-${Math.random().toString(36).substring(2, 7)}`,
          toolName: call.toolName,
          arguments: call.arguments,
          status: 'running',
          dispatchedAt: Date.now(),
        };

        this.state.toolRegistry.set(deduplicationKey, toolSpec);

        // Eksekusi background asinkron
        this.toolExecutor(call.toolName, call.arguments, this.currentAbortController.signal)
          .then((res) => {
            toolSpec.status = 'completed';
            toolSpec.result = res;
            toolSpec.completedAt = Date.now();
          })
          .catch((err) => {
            if (toolSpec.status !== 'cancelled') {
              toolSpec.status = 'completed';
              toolSpec.result = { error: String(err) };
            }
          });
      }
    } finally {
      this.state.activeThinkerCount = Math.max(0, this.state.activeThinkerCount - 1);
    }
  }

  // Transisi Endpoint: Dipicu oleh Voice Activity Detector (VAD) saat ucapan selesai
  public handleVADEndpointHandoff(finalTranscript: string): {
    finalTranscript: string;
    completedPreparation: {
      reasoning: ReasoningRecord[];
      usableToolResults: Array<{ toolName: string; args: Record; result: unknown }>;
    };
  } {
    // 1. Batalkan semua task spekulasi yang masih menggantung
    if (this.currentAbortController) {
      this.currentAbortController.abort();
    }

    // 2. Kumpulkan artefak yang telah selesai sebelum t_endpoint
    const usableToolResults: Array<{ toolName: string; args: Record; result: unknown }> = [];
    for (const [key, spec] of this.state.toolRegistry.entries()) {
      if (spec.status === 'completed' && spec.result !== undefined) {
        usableToolResults.push({
          toolName: spec.toolName,
          args: spec.arguments,
          result: spec.result,
        });
      } else {
        spec.status = 'cancelled';
      }
    }

    // 3. Serahkan langsung ke Responder context
    return {
      finalTranscript,
      completedPreparation: {
        reasoning: [...this.state.completedReasoning],
        usableToolResults,
      },
    };
  }

  private async evaluateThinkerStep(
    id: number,
    transcript: string,
    completedTools: unknown[],
    signal: AbortSignal
  ): Promise<{ reasoning?: string; requestedTools: Array<{ toolName: string; arguments: Record }> }> {
    // Stub implementasi LLM Thinker invocation
    return { requestedTools: [] };
  }
}