From 34d5d5e88e4b58841573d3a58ee527a60b10c532 Mon Sep 17 00:00:00 2001
From: Hong Jiarong <me@jrhim.com>
Date: Mon, 27 Jul 2026 12:25:04 +0800
Subject: [PATCH] fix(hub): do not crash Hub on missing DocMind input files

createReadStream emits async ENOENT without a listener, which became an
unhandled 'error' event and exited the silo process. Teachers then saw the
startup "process restart" notice. Wait for stream open and convert missing
files into DocmindClientError instead.
---
 hub/src/capability/docmindClient.ts  | 42 ++++++++++++++++++++++++----
 hub/test/unit/docmind-client.test.ts | 34 ++++++++++++++++++++++
 2 files changed, 70 insertions(+), 6 deletions(-)
 create mode 100644 hub/test/unit/docmind-client.test.ts

diff --git a/hub/src/capability/docmindClient.ts b/hub/src/capability/docmindClient.ts
index 1e9a3b7..716febd 100644
--- a/hub/src/capability/docmindClient.ts
+++ b/hub/src/capability/docmindClient.ts
@@ -19,7 +19,8 @@ import $DocmindClient, {
   QueryDocParserStatusRequest,
 } from "@alicloud/docmind-api20220711";
 import { RuntimeOptions } from "@alicloud/tea-util";
-import { createReadStream } from "node:fs";
+import { createReadStream, type ReadStream } from "node:fs";
+import { once } from "node:events";
 import { basename } from "node:path";
 import type { DocmindCapabilitySecretPayload } from "./types.js";
 
@@ -66,6 +67,11 @@ type DocmindConfig = ConstructorParameters<typeof $DocmindClient.default>[0];
 
 export class AliyunDocmindClient implements CapabilityProviderClient {
   async parse(credential: DocmindCapabilitySecretPayload, options: DocmindParseOptions): Promise<DocmindParseResult> {
+    // Open first so missing local inputs fail closed before touching the SDK.
+    // Unhandled createReadStream('error') previously crashed the Hub process.
+    const fileName = basename(options.inputFilePath);
+    const fileStream = await openLocalFileStream(options.inputFilePath);
+
     const config: DocmindConfig = {
       endpoint: credential.endpoint,
       accessKeyId: credential.accessKeyId,
@@ -75,11 +81,9 @@ export class AliyunDocmindClient implements CapabilityProviderClient {
     } as DocmindConfig;
     const client = new $DocmindClient.default(config);
 
-    // 1. Submit job with local file as a ReadStream (not a Buffer — the SDK
-    //    serializes Buffers as JSON {type:"Buffer",data:[...]} which the API
-    //    can't read; a Stream is uploaded as multipart form data).
-    const fileName = basename(options.inputFilePath);
-    const fileStream = createReadStream(options.inputFilePath);
+    // Submit job with local file as a ReadStream (not a Buffer — the SDK
+    // serializes Buffers as JSON {type:"Buffer",data:[...]} which the API
+    // can't read; a Stream is uploaded as multipart form data).
     const advanceRequest = new SubmitDocParserJobAdvanceRequest({
       fileUrlObject: fileStream,
       fileName,
@@ -91,6 +95,7 @@ export class AliyunDocmindClient implements CapabilityProviderClient {
     try {
       submitResponse = await client.submitDocParserJobAdvance(advanceRequest, runtime);
     } catch (e) {
+      fileStream.destroy();
       throw new DocmindClientError(
         e instanceof Error ? e.message : String(e),
         "docmind_unreachable",
@@ -229,3 +234,28 @@ function extractFilename(altText: string, url: string, index: number): string {
   if (base !== "" && base !== "/") return base;
   return `image_${index + 1}.png`;
 }
+
+/**
+ * Open a local file as a ReadStream only after the fd is successfully open.
+ * createReadStream() emits asynchronous 'error' for missing paths; without a
+ * listener that becomes an unhandled EventEmitter error and exits Node.
+ */
+async function openLocalFileStream(path: string): Promise<ReadStream> {
+  const stream = createReadStream(path);
+  try {
+    await once(stream, "open");
+  } catch (error) {
+    stream.destroy();
+    const err = error instanceof Error ? error : new Error(String(error));
+    const code = (err as NodeJS.ErrnoException).code;
+    if (code === "ENOENT") {
+      throw new DocmindClientError(`input file not found: ${path}`, "docmind_rejected");
+    }
+    throw new DocmindClientError(err.message, "docmind_unreachable");
+  }
+  // After open, residual stream errors must not become unhandled and crash Hub.
+  stream.on("error", () => {
+    // The Aliyun SDK / destroy path owns consumption failures after open.
+  });
+  return stream;
+}
diff --git a/hub/test/unit/docmind-client.test.ts b/hub/test/unit/docmind-client.test.ts
new file mode 100644
index 0000000..ac476a0
--- /dev/null
+++ b/hub/test/unit/docmind-client.test.ts
@@ -0,0 +1,34 @@
+import { join } from "node:path";
+import { tmpdir } from "node:os";
+import { describe, expect, it } from "vitest";
+import { AliyunDocmindClient, DocmindClientError } from "../../src/capability/docmindClient.js";
+
+describe("AliyunDocmindClient local file open", () => {
+  it("rejects a missing input file without crashing the process", async () => {
+    const client = new AliyunDocmindClient();
+    const missing = join(tmpdir(), `docmind-missing-${Date.now()}.pdf`);
+
+    // If createReadStream errors are left unhandled, Vitest aborts the suite
+    // with an unhandled 'error' event instead of reaching this assertion.
+    await expect(client.parse(
+      {
+        accessKeyId: "ak",
+        accessKeySecret: "sk",
+        endpoint: "docmind-api.cn-hangzhou.aliyuncs.com",
+      },
+      { inputFilePath: missing },
+    )).rejects.toBeInstanceOf(DocmindClientError);
+
+    await expect(client.parse(
+      {
+        accessKeyId: "ak",
+        accessKeySecret: "sk",
+        endpoint: "docmind-api.cn-hangzhou.aliyuncs.com",
+      },
+      { inputFilePath: missing },
+    )).rejects.toMatchObject({
+      code: "docmind_rejected",
+      message: expect.stringContaining("input file not found"),
+    });
+  });
+});