fix(dictation): avoid timeout from dangling non-final segments

2026-02-18 21:21:27 +07:00
parent bcae95c372
commit a1fdeeb7ae
16 changed files with 419 additions and 10 deletions
@@ -58,6 +58,7 @@ import {
 import { loadOrCreateDaemonKeyPair } from "./daemon-keypair.js";
 import { startRelayTransport, type RelayTransportController } from "./relay-transport.js";
 import { getOrCreateServerId } from "./server-id.js";
+import { resolveDaemonVersion } from "./daemon-version.js";
 import type {
  AgentClient,
  AgentProvider,
@@ -136,6 +137,7 @@ export async function createPaseoDaemon(
  rootLogger: Logger
 ): Promise<PaseoDaemon> {
  const logger = rootLogger.child({ module: "bootstrap" });
+  const daemonVersion = resolveDaemonVersion(import.meta.url);

  // Acquire PID lock before expensive bootstrap work so duplicate starts fail immediately.
  await acquirePidLock(config.paseoHome, config.listen);
@@ -496,7 +498,8 @@ export async function createPaseoDaemon(
      localModels: localModelConfig ?? undefined,
      getSpeechReadiness,
    },
-    config.agentProviderSettings
+    config.agentProviderSettings,
+    daemonVersion
  );
  unsubscribeSpeechReadiness = subscribeSpeechReadiness((snapshot) => {
    wsServer?.publishSpeechReadiness(snapshot);
@@ -0,0 +1,71 @@
+import { afterEach, describe, expect, it } from "vitest";
+import { mkdirSync, mkdtempSync, rmSync, writeFileSync } from "node:fs";
+import { tmpdir } from "node:os";
+import path from "node:path";
+import { pathToFileURL } from "node:url";
+import {
+  DaemonVersionResolutionError,
+  resolveDaemonVersion,
+} from "./daemon-version.js";
+
+const createdDirs: string[] = [];
+
+function createTempDir(): string {
+  const dir = mkdtempSync(path.join(tmpdir(), "paseo-daemon-version-"));
+  createdDirs.push(dir);
+  return dir;
+}
+
+afterEach(() => {
+  for (const dir of createdDirs.splice(0, createdDirs.length)) {
+    rmSync(dir, { recursive: true, force: true });
+  }
+});
+
+describe("resolveDaemonVersion", () => {
+  it("resolves server version by walking up to @getpaseo/server package.json", () => {
+    const root = createTempDir();
+    writeFileSync(
+      path.join(root, "package.json"),
+      JSON.stringify({ name: "@getpaseo/server", version: "9.8.7" }),
+      "utf8"
+    );
+    const nestedDir = path.join(root, "dist", "server");
+    mkdirSync(nestedDir, { recursive: true });
+
+    const moduleUrl = pathToFileURL(path.join(nestedDir, "index.js")).href;
+    expect(resolveDaemonVersion(moduleUrl)).toBe("9.8.7");
+  });
+
+  it("throws when @getpaseo/server package metadata cannot be resolved", () => {
+    const root = createTempDir();
+    writeFileSync(
+      path.join(root, "package.json"),
+      JSON.stringify({ name: "not-getpaseo-server", version: "1.2.3" }),
+      "utf8"
+    );
+    const nestedDir = path.join(root, "dist", "server");
+    mkdirSync(nestedDir, { recursive: true });
+
+    const moduleUrl = pathToFileURL(path.join(nestedDir, "index.js")).href;
+    expect(() => resolveDaemonVersion(moduleUrl)).toThrow(
+      DaemonVersionResolutionError
+    );
+  });
+
+  it("throws when @getpaseo/server version is missing", () => {
+    const root = createTempDir();
+    writeFileSync(
+      path.join(root, "package.json"),
+      JSON.stringify({ name: "@getpaseo/server" }),
+      "utf8"
+    );
+    const nestedDir = path.join(root, "dist", "server");
+    mkdirSync(nestedDir, { recursive: true });
+
+    const moduleUrl = pathToFileURL(path.join(nestedDir, "index.js")).href;
+    expect(() => resolveDaemonVersion(moduleUrl)).toThrow(
+      DaemonVersionResolutionError
+    );
+  });
+});
@@ -0,0 +1,25 @@
+import {
+  PackageVersionResolutionError,
+  resolvePackageVersion,
+} from "./package-version.js";
+
+const SERVER_PACKAGE_NAME = "@getpaseo/server";
+
+export class DaemonVersionResolutionError extends PackageVersionResolutionError {}
+
+export function resolveDaemonVersion(moduleUrl: string = import.meta.url): string {
+  try {
+    return resolvePackageVersion({
+      moduleUrl,
+      packageName: SERVER_PACKAGE_NAME,
+    });
+  } catch (error) {
+    if (error instanceof PackageVersionResolutionError) {
+      throw new DaemonVersionResolutionError({
+        moduleUrl,
+        packageName: SERVER_PACKAGE_NAME,
+      });
+    }
+    throw error;
+  }
+}
@@ -233,4 +233,82 @@ describe("DictationStreamManager (provider-agnostic provider)", () => {
    expect(finishAccepted).toBeDefined();
    expect(finishAccepted?.payload.timeoutMs).toBeGreaterThan(5000);
  });
+
+  it("adapts finish timeout when only uncommitted non-final transcripts are pending", async () => {
+    const session = new FakeRealtimeSession();
+    const emitted: Array<{ type: string; payload: any }> = [];
+    const manager = new DictationStreamManager({
+      logger: pino({ level: "silent" }),
+      emit: (msg) => emitted.push(msg),
+      sessionId: "s1",
+      stt: new FakeSttProvider(session),
+      finalTimeoutMs: 5000,
+    });
+
+    await manager.handleStart("d-uncommitted-timeout", "audio/pcm;rate=24000;bits=16");
+    await manager.handleChunk({
+      dictationId: "d-uncommitted-timeout",
+      seq: 0,
+      audioBase64: buildPcmBase64(2000, 2400),
+      format: "audio/pcm;rate=24000;bits=16",
+    });
+
+    session.emitCommitted("seg-1");
+    session.emitTranscript("seg-1", "hello", true);
+    session.emitTranscript("seg-dangling", "hel", false);
+
+    await manager.handleFinish("d-uncommitted-timeout", 0);
+
+    const finishAccepted = emitted.find(
+      (msg) => msg.type === "dictation_stream_finish_accepted"
+    );
+    expect(finishAccepted).toBeDefined();
+    expect(finishAccepted?.payload.timeoutMs).toBeGreaterThan(5000);
+  });
+
+  it("drops dangling uncommitted non-final transcripts when finishing after silence tail clear", async () => {
+    vi.useFakeTimers();
+    try {
+      const session = new FakeRealtimeSession();
+      const emitted: Array<{ type: string; payload: any }> = [];
+      const manager = new DictationStreamManager({
+        logger: pino({ level: "silent" }),
+        emit: (msg) => emitted.push(msg),
+        sessionId: "s1",
+        stt: new FakeSttProvider(session),
+        finalTimeoutMs: 5000,
+      });
+
+      await manager.handleStart("d-clear-tail", "audio/pcm;rate=24000;bits=16");
+      await manager.handleChunk({
+        dictationId: "d-clear-tail",
+        seq: 0,
+        audioBase64: buildPcmBase64(2000, 2400),
+        format: "audio/pcm;rate=24000;bits=16",
+      });
+
+      session.emitCommitted("seg-1");
+      session.emitTranscript("seg-1", "hello", true);
+
+      await manager.handleChunk({
+        dictationId: "d-clear-tail",
+        seq: 1,
+        audioBase64: buildPcmBase64(0, 2400),
+        format: "audio/pcm;rate=24000;bits=16",
+      });
+      session.emitTranscript("seg-dangling", "", false);
+
+      await manager.handleFinish("d-clear-tail", 1);
+      await tick();
+      await vi.advanceTimersByTimeAsync(5_100);
+
+      const final = emitted.find((msg) => msg.type === "dictation_stream_final");
+      const error = emitted.find((msg) => msg.type === "dictation_stream_error");
+      expect(session.clearCalls).toBeGreaterThan(0);
+      expect(error).toBeUndefined();
+      expect(final?.payload.text).toBe("hello");
+    } finally {
+      vi.useRealTimers();
+    }
+  });
 });
@@ -552,8 +552,19 @@ export class DictationStreamManager {
    const pendingCommittedSegments = state.committedSegmentIds.reduce((count, segmentId) => {
      return state.finalTranscriptSegmentIds.has(segmentId) ? count : count + 1;
    }, 0);
+    const committedSet = new Set(state.committedSegmentIds);
+    const pendingUncommittedTranscriptSegments = Array.from(
+      state.transcriptsBySegmentId.keys()
+    ).reduce((count, segmentId) => {
+      if (committedSet.has(segmentId)) {
+        return count;
+      }
+      return state.finalTranscriptSegmentIds.has(segmentId) ? count : count + 1;
+    }, 0);
    const pendingSegments =
-      pendingCommittedSegments + (state.awaitingFinalCommit ? 1 : 0);
+      pendingCommittedSegments +
+      pendingUncommittedTranscriptSegments +
+      (state.awaitingFinalCommit ? 1 : 0);
    const pendingAudioSeconds = Math.ceil(
      Math.max(0, state.bytesSinceCommit) / bytesPerSecond
    );
@@ -629,6 +640,16 @@ export class DictationStreamManager {
        state.bytesSinceCommit = 0;
        state.peakSinceCommit = 0;
        state.awaitingFinalCommit = false;
+        const droppedSegments = this.dropUncommittedNonFinalTranscripts(state);
+        if (droppedSegments > 0) {
+          this.logger.debug(
+            {
+              dictationId,
+              droppedSegments,
+            },
+            "Dictation finish: dropped uncommitted non-final transcript segments after silence clear"
+          );
+        }
      } else {
        state.awaitingFinalCommit = true;
        try {
@@ -646,6 +667,22 @@ export class DictationStreamManager {
    state.finishSealed = true;
  }

+  private dropUncommittedNonFinalTranscripts(state: DictationStreamState): number {
+    const committedSet = new Set(state.committedSegmentIds);
+    let droppedCount = 0;
+    for (const segmentId of state.transcriptsBySegmentId.keys()) {
+      if (committedSet.has(segmentId)) {
+        continue;
+      }
+      if (state.finalTranscriptSegmentIds.has(segmentId)) {
+        continue;
+      }
+      state.transcriptsBySegmentId.delete(segmentId);
+      droppedCount += 1;
+    }
+    return droppedCount;
+  }
+
  private maybeFinalizeDictationStream(dictationId: string): void {
    const state = this.streams.get(dictationId);
    if (!state) {
@@ -0,0 +1,64 @@
+import { existsSync, readFileSync } from "node:fs";
+import path from "node:path";
+import { fileURLToPath } from "node:url";
+
+type ResolvePackageVersionParams = {
+  moduleUrl?: string;
+  packageName: string;
+};
+
+type PackageJson = {
+  name?: unknown;
+  version?: unknown;
+};
+
+export class PackageVersionResolutionError extends Error {
+  constructor(params: { moduleUrl: string; packageName: string }) {
+    super(
+      `Unable to resolve ${params.packageName} version from module URL ${params.moduleUrl}.`
+    );
+    this.name = "PackageVersionResolutionError";
+  }
+}
+
+export function resolvePackageVersion(params: ResolvePackageVersionParams): string {
+  const moduleUrl = params.moduleUrl ?? import.meta.url;
+  let currentDir = path.dirname(fileURLToPath(moduleUrl));
+
+  while (true) {
+    const packageJsonPath = path.join(currentDir, "package.json");
+    if (existsSync(packageJsonPath)) {
+      try {
+        const packageJson = JSON.parse(readFileSync(packageJsonPath, "utf8")) as PackageJson;
+        if (packageJson.name === params.packageName) {
+          if (
+            typeof packageJson.version === "string" &&
+            packageJson.version.trim().length > 0
+          ) {
+            return packageJson.version.trim();
+          }
+          throw new PackageVersionResolutionError({
+            moduleUrl,
+            packageName: params.packageName,
+          });
+        }
+      } catch (error) {
+        if (error instanceof PackageVersionResolutionError) {
+          throw error;
+        }
+        // Continue searching parent directories.
+      }
+    }
+
+    const parentDir = path.dirname(currentDir);
+    if (parentDir === currentDir) {
+      break;
+    }
+    currentDir = parentDir;
+  }
+
+  throw new PackageVersionResolutionError({
+    moduleUrl,
+    packageName: params.packageName,
+  });
+}
@@ -75,7 +75,13 @@ function createServer(agentManagerOverrides?: Record<string, unknown>) {
    {} as any,
    "/tmp/paseo-test",
    async () => ({} as any),
-    { allowedOrigins: new Set() }
+    { allowedOrigins: new Set() },
+    undefined,
+    undefined,
+    undefined,
+    undefined,
+    undefined,
+    "1.2.3-test"
  );

  return { server, agentManager };
@@ -79,6 +79,8 @@ import {
 import { parseServerInfoStatusPayload } from "./messages.js";
 import type { SpeechReadinessSnapshot } from "./speech/speech-runtime.js";

+const TEST_DAEMON_VERSION = "1.2.3-test";
+
 class MockSocket {
  readyState = 1;
  sent: unknown[] = [];
@@ -157,7 +159,9 @@ function createServer(options?: { speechReadiness?: SpeechReadinessSnapshot | nu
      ? {
          getSpeechReadiness: () => speechReadiness,
        }
-      : undefined
+      : undefined,
+    undefined,
+    TEST_DAEMON_VERSION
  );
 }

@@ -318,6 +322,7 @@ describe("relay external socket reconnect behavior", () => {

    const payload = parseServerInfoStatusPayload(envelope.message?.payload);
    expect(payload?.status).toBe("server_info");
+    expect(payload?.version).toBe(TEST_DAEMON_VERSION);
    expect(payload?.capabilities?.voice?.dictation.enabled).toBe(
      speechReadiness.dictation.enabled
    );
@@ -158,6 +158,13 @@ type SessionConnection = {

 const EXTERNAL_SESSION_DISCONNECT_GRACE_MS = 90_000;

+export class MissingDaemonVersionError extends Error {
+  constructor() {
+    super("VoiceAssistantWebSocketServer requires a non-empty daemonVersion.");
+    this.name = "MissingDaemonVersionError";
+  }
+}
+
 /**
 * WebSocket server that only accepts sockets + parses/forwards messages to the session layer.
 */
@@ -168,6 +175,7 @@ export class VoiceAssistantWebSocketServer {
  private readonly externalSessionsByKey: Map<string, SessionConnection> = new Map();
  private clientIdCounter = 0;
  private readonly serverId: string;
+  private readonly daemonVersion: string;
  private readonly agentManager: AgentManager;
  private readonly agentStorage: AgentStorage;
  private readonly downloadTokenStore: DownloadTokenStore;
@@ -229,10 +237,15 @@ export class VoiceAssistantWebSocketServer {
      };
      getSpeechReadiness?: () => SpeechReadinessSnapshot;
    },
-    agentProviderRuntimeSettings?: AgentProviderRuntimeSettingsMap
+    agentProviderRuntimeSettings?: AgentProviderRuntimeSettingsMap,
+    daemonVersion?: string
  ) {
    this.logger = logger.child({ module: "websocket-server" });
    this.serverId = serverId;
+    if (typeof daemonVersion !== "string" || daemonVersion.trim().length === 0) {
+      throw new MissingDaemonVersionError();
+    }
+    this.daemonVersion = daemonVersion.trim();
    this.agentManager = agentManager;
    this.agentStorage = agentStorage;
    this.downloadTokenStore = downloadTokenStore;
@@ -509,6 +522,7 @@ export class VoiceAssistantWebSocketServer {
      status: "server_info",
      serverId: this.serverId,
      hostname: getHostname(),
+      version: this.daemonVersion,
      ...(this.serverCapabilities ? { capabilities: this.serverCapabilities } : {}),
    };
  }
@@ -1218,6 +1218,14 @@ const ServerInfoHostnameSchema = z.unknown().transform((value): string | null =>
  return trimmed.length > 0 ? trimmed : null
 })

+const ServerInfoVersionSchema = z.unknown().transform((value): string | null => {
+  if (typeof value !== 'string') {
+    return null
+  }
+  const trimmed = value.trim()
+  return trimmed.length > 0 ? trimmed : null
+})
+
 const ServerCapabilitiesFromUnknownSchema = z
  .unknown()
  .optional()
@@ -1237,12 +1245,14 @@ export const ServerInfoStatusPayloadSchema = z
    status: z.literal('server_info'),
    serverId: z.string().trim().min(1),
    hostname: ServerInfoHostnameSchema.optional(),
+    version: ServerInfoVersionSchema.optional(),
    capabilities: ServerCapabilitiesFromUnknownSchema,
  })
  .passthrough()
  .transform((payload) => ({
    ...payload,
    hostname: payload.hostname ?? null,
+    version: payload.version ?? null,
  }))

 export const StatusMessageSchema = z.object({
@@ -8,10 +8,15 @@ import {
  computeExecutionOrder,
  buildSortedChildrenMap,
 } from "./execution-order.js";
+import { resolvePackageVersion } from "../server/package-version.js";
 import type { AgentType, ModelName, Task } from "./types.js";

 const TASKS_DIR = resolve(process.cwd(), ".tasks");
 const store = new FileTaskStore(TASKS_DIR);
+const TASK_CLI_VERSION = resolvePackageVersion({
+  moduleUrl: import.meta.url,
+  packageName: "@getpaseo/server",
+});

 async function readStdin(): Promise<string> {
  const chunks: Buffer[] = [];
@@ -24,7 +29,7 @@ async function readStdin(): Promise<string> {
 const program = new Command()
  .name("task")
  .description("Minimal task management with dependency tracking")
-  .version("0.1.0")
+  .version(TASK_CLI_VERSION)
  .addHelpText(
    "after",
    `