{"version":3,"file":"queue-adapter-agent-server.cjs","names":["#runs","#assistantId","#store","#onError","#getThread","#threadId","#ensureWatching","#hydrateGeneration","#knownRunIds","#abortController","#isStillQueued","#toEntry","#startedEventCount","bindThreadConfig","#pendingCancel","#refreshPending","#cancelEntry","EMPTY_QUEUE","#unsubscribe","#refreshRetryTimer"],"sources":["../../src/stream/queue-adapter-agent-server.ts"],"sourcesContent":["import { v7 as uuidv7 } from \"@langchain/core/utils/uuid\";\nimport type { ThreadStream } from \"../client/index.js\";\nimport type { Run } from \"../schema.js\";\nimport { bindThreadConfig } from \"./dispatch-config.js\";\nimport {\n  EMPTY_QUEUE,\n  type QueueAdapter,\n  type ServerQueueCapability,\n  type SubmissionQueueEntry,\n  type SubmissionQueueSnapshot,\n} from \"./queue-adapter.js\";\nimport { StreamStore } from \"./store.js\";\nimport type { StreamSubmitOptions } from \"./types.js\";\n\n/** `runs.list()`'s `Run` type doesn't declare `kwargs` by default; it's an opt-in `select` field. */\ntype RunWithKwargs = Run & { kwargs?: { input?: unknown } };\n\n/**\n * Backs \"enqueue\" with real, durable runs. An entry is a genuine\n * server-side pending run the instant `enqueue()` resolves.\n *\n * Selected when `queue: \"server\"` is set (see {@link SubmitCoordinator}'s\n * constructor).\n */\nexport class AgentServerQueueAdapter<\n  StateType extends object = Record<string, unknown>,\n> implements QueueAdapter<StateType> {\n  readonly #runs: ServerQueueCapability;\n  readonly #assistantId: string;\n  readonly #store: StreamStore<SubmissionQueueSnapshot<StateType>>;\n  readonly #onError: (error: unknown) => void;\n  readonly #getThread: (threadId: string) => Pick<ThreadStream, \"onEvent\">;\n  /** One shared subscription for this adapter's lifetime, not one per run. */\n  #unsubscribe: (() => void) | undefined;\n  /** ids canceled while their create() was still in flight. */\n  #pendingCancel = new Set<string>();\n  #threadId: string | undefined;\n  /** Guards against a stale hydrate() landing after a rapid thread-switch. */\n  #hydrateGeneration = 0;\n  /** Aborts in-flight bookkeeping `list()` calls on detach(); replaced (never reused) afterward. */\n  #abortController = new AbortController();\n  /** Handle for #refreshPending's pending retry, if any; cleared on detach(). */\n  #refreshRetryTimer: ReturnType<typeof setTimeout> | undefined;\n  /** Lets enqueue() detect a \"started\" event during its own create(). */\n  #startedEventCount = 0;\n\n  constructor(\n    runs: ServerQueueCapability,\n    assistantId: string,\n    store: StreamStore<SubmissionQueueSnapshot<StateType>>,\n    onError: (error: unknown) => void,\n    getThread: (threadId: string) => Pick<ThreadStream, \"onEvent\">\n  ) {\n    this.#runs = runs;\n    this.#assistantId = assistantId;\n    this.#store = store;\n    this.#onError = onError;\n    this.#getThread = getThread;\n  }\n\n  async hydrate(threadId: string): Promise<void> {\n    this.#threadId = threadId;\n    this.#ensureWatching(threadId);\n    const generation = ++this.#hydrateGeneration;\n    // Baseline before asking: an entry assigned after this can't be in the response below.\n    const knownAtRequestTime = this.#knownRunIds();\n    let pending: RunWithKwargs[];\n    try {\n      pending = (await this.#runs.list(threadId, {\n        status: \"pending\",\n        // `list()` defaults to 10; a real cap would drop older entries.\n        limit: 1000,\n        // \"kwargs\" isn't part of the base `Run` shape returned by list();\n        // it must be explicitly selected to get the original input back.\n        select: [\"run_id\", \"kwargs\", \"created_at\", \"multitask_strategy\"],\n        signal: this.#abortController.signal,\n      })) as RunWithKwargs[];\n    } catch (error) {\n      // A generation mismatch means detach() aborted this on purpose.\n      if (generation === this.#hydrateGeneration) this.#onError(error);\n      return;\n    }\n    if (generation !== this.#hydrateGeneration) return; // superseded by a later hydrate()/detach()\n\n    const remoteIds = new Set(pending.map((run) => run.run_id));\n    this.#store.setState((current) => {\n      const knownRunIds = new Set(\n        current.flatMap((e) => (e.runId ? [e.runId] : []))\n      );\n      // Local entries are never overwritten by a server reconstruction\n      // — a consumer may already be keying UI on their id.\n      const stillValid = current.filter((e) =>\n        this.#isStillQueued(e, remoteIds, knownAtRequestTime)\n      );\n      const newFromRemote = pending\n        .filter((run) => !knownRunIds.has(run.run_id))\n        .map((run) => this.#toEntry(run));\n      return [\n        ...newFromRemote,\n        ...stillValid,\n      ] as SubmissionQueueSnapshot<StateType>;\n    });\n  }\n\n  /**\n   * On a self-created thread, this can race the active run's own\n   * `run.start` request that creates the thread row: if this\n   * `runs.create()` reaches the server first, it may 404. The local\n   * adapter has no such race (it never hits the server while queued).\n   */\n  async enqueue(\n    threadId: string,\n    values: Partial<StateType> | null | undefined,\n    options: StreamSubmitOptions<StateType> | undefined\n  ): Promise<void> {\n    // A brand-new self-created thread may never be hydrated first.\n    this.#threadId = threadId;\n    this.#ensureWatching(threadId);\n    const startedEventsBefore = this.#startedEventCount;\n    const generationAtEnqueue = this.#hydrateGeneration;\n    const id = uuidv7();\n    this.#store.setState((current) => [\n      ...current,\n      { id, values, options, createdAt: new Date() },\n    ]);\n    try {\n      const run = await this.#runs.create(threadId, this.#assistantId, {\n        input: values,\n        config: bindThreadConfig(options?.config, threadId),\n        metadata: options?.metadata,\n        checkpointId: options?.forkFrom,\n        multitaskStrategy: \"enqueue\",\n      });\n      if (this.#pendingCancel.delete(id)) {\n        await this.#runs.cancel(threadId, run.run_id); // canceled while creating; don't resurrect it\n        return;\n      }\n      this.#store.setState((current) =>\n        current.map((entry) =>\n          entry.id === id ? { ...entry, runId: run.run_id } : entry\n        )\n      );\n      // A run may have started before we learned its id; re-check now.\n      // Skip if detach()/hydrate() rebound the adapter to another thread\n      // in the meantime — `threadId` here is this call's own thread, not\n      // necessarily the one the adapter is currently watching.\n      if (\n        this.#startedEventCount !== startedEventsBefore &&\n        this.#hydrateGeneration === generationAtEnqueue\n      ) {\n        void this.#refreshPending(threadId);\n      }\n    } catch (error) {\n      this.#pendingCancel.delete(id); // avoid leaking a pending-cancel marker nothing will ever consult\n      this.#store.setState((current) =>\n        current.filter((entry) => entry.id !== id)\n      );\n      this.#onError(error);\n      throw error;\n    }\n  }\n\n  async cancel(id: string): Promise<boolean> {\n    const entry = this.#store.getSnapshot().find((e) => e.id === id);\n    if (!entry) return false;\n    this.#store.setState((current) => current.filter((e) => e.id !== id));\n    await this.#cancelEntry(entry);\n    return true;\n  }\n\n  async clear(): Promise<void> {\n    const entries = this.#store.getSnapshot();\n    this.#store.setState(\n      () => EMPTY_QUEUE as SubmissionQueueSnapshot<StateType>\n    );\n    // `allSettled`, not `all`: one failed cancel must not stop the rest\n    // from being attempted, nor reject `clear()` itself.\n    const results = await Promise.allSettled(\n      entries.map((entry) => this.#cancelEntry(entry))\n    );\n    const failures = results.filter(\n      (r): r is PromiseRejectedResult => r.status === \"rejected\"\n    );\n    if (failures.length === 1) {\n      this.#onError(failures[0].reason);\n    } else if (failures.length > 1) {\n      this.#onError(\n        new AggregateError(\n          failures.map((f) => f.reason),\n          `${failures.length} of ${entries.length} queued runs failed to cancel`\n        )\n      );\n    }\n  }\n\n  detach(): void {\n    this.#hydrateGeneration++; // invalidate any in-flight hydrate() for this thread\n    this.#unsubscribe?.();\n    this.#unsubscribe = undefined;\n    // #pendingCancel is not cleared: dropping it would silently un-cancel an in-flight create().\n    this.#threadId = undefined;\n    this.#abortController.abort(); // stop any in-flight hydrate()/#refreshPending() list() call\n    this.#abortController = new AbortController(); // an aborted controller can't be reused\n    clearTimeout(this.#refreshRetryTimer); // a scheduled retry outliving detach() would target a stale thread\n    this.#refreshRetryTimer = undefined;\n  }\n\n  async #cancelEntry(entry: SubmissionQueueEntry<StateType>): Promise<void> {\n    if (entry.runId && this.#threadId) {\n      await this.#runs.cancel(this.#threadId, entry.runId); // already real, idempotent\n    } else {\n      this.#pendingCancel.add(entry.id); // still being created; cancel on arrival\n    }\n  }\n\n  /**\n   * Lazily opens ONE subscription for this adapter's lifetime, not one\n   * per run. `LifecycleEvent` carries no `run_id`, so a `\"running\"`\n   * event triggers a `runs.list` re-check instead of a direct match.\n   */\n  #ensureWatching(threadId: string): void {\n    if (this.#unsubscribe) return;\n    this.#unsubscribe = this.#getThread(threadId).onEvent((event) => {\n      if (event.method !== \"lifecycle\") return;\n      if (event.params.data.event !== \"running\") return;\n      this.#startedEventCount++;\n      void this.#refreshPending(threadId);\n    });\n  }\n\n  /**\n   * Re-lists pending runs and drops any queue entry no longer reported\n   * pending: it's running now. Retried once after a network failure\n   * before surfacing an error — otherwise a single blip would leave a\n   * now-running entry stuck showing \"queued\" until the next promotion\n   * or a manual hydrate.\n   */\n  async #refreshPending(threadId: string, attempt = 0): Promise<void> {\n    if (this.#store.getSnapshot().length === 0) return; // nothing left to refresh\n    const generation = this.#hydrateGeneration;\n    const knownAtRequestTime = this.#knownRunIds(); // same reasoning as hydrate()\n    try {\n      const pending = await this.#runs.list(threadId, {\n        status: \"pending\",\n        limit: 1000, // list() defaults to 10; see the matching note in hydrate()\n        signal: this.#abortController.signal,\n      });\n      if (generation !== this.#hydrateGeneration) return; // superseded; thread id is stale\n      const stillPendingIds = new Set(pending.map((run) => run.run_id));\n      this.#store.setState((current) =>\n        current.filter((entry) =>\n          this.#isStillQueued(entry, stillPendingIds, knownAtRequestTime)\n        )\n      );\n    } catch (error) {\n      if (generation !== this.#hydrateGeneration) return; // detach() aborted this on purpose\n      if (attempt < 1) {\n        this.#refreshRetryTimer = setTimeout(() => {\n          this.#refreshRetryTimer = undefined;\n          void this.#refreshPending(threadId, attempt + 1);\n        }, 2000);\n        return;\n      }\n      this.#onError(error);\n    }\n  }\n\n  #toEntry(run: RunWithKwargs): SubmissionQueueEntry<StateType> {\n    return {\n      id: run.run_id,\n      runId: run.run_id,\n      values: run.kwargs?.input as Partial<StateType> | null | undefined,\n      createdAt: new Date(run.created_at),\n    };\n  }\n\n  #knownRunIds(): Set<string> {\n    return new Set(\n      this.#store.getSnapshot().flatMap((e) => (e.runId ? [e.runId] : []))\n    );\n  }\n\n  /** Shared by hydrate() and #refreshPending(): is a list() response grounds to evict this entry? */\n  #isStillQueued(\n    entry: SubmissionQueueEntry<StateType>,\n    stillPendingIds: Set<string>,\n    knownAtRequestTime: Set<string>\n  ): boolean {\n    return (\n      !entry.runId ||\n      stillPendingIds.has(entry.runId) ||\n      !knownAtRequestTime.has(entry.runId)\n    );\n  }\n}\n"],"mappings":";;;;;;;;;;;;AAwBA,IAAa,0BAAb,MAEqC;CACnC;CACA;CACA;CACA;CACA;;CAEA;;CAEA,iCAAiB,IAAI,KAAa;CAClC;;CAEA,qBAAqB;;CAErB,mBAAmB,IAAI,iBAAiB;;CAExC;;CAEA,qBAAqB;CAErB,YACE,MACA,aACA,OACA,SACA,WACA;AACA,QAAA,OAAa;AACb,QAAA,cAAoB;AACpB,QAAA,QAAc;AACd,QAAA,UAAgB;AAChB,QAAA,YAAkB;;CAGpB,MAAM,QAAQ,UAAiC;AAC7C,QAAA,WAAiB;AACjB,QAAA,eAAqB,SAAS;EAC9B,MAAM,aAAa,EAAE,MAAA;EAErB,MAAM,qBAAqB,MAAA,aAAmB;EAC9C,IAAI;AACJ,MAAI;AACF,aAAW,MAAM,MAAA,KAAW,KAAK,UAAU;IACzC,QAAQ;IAER,OAAO;IAGP,QAAQ;KAAC;KAAU;KAAU;KAAc;KAAqB;IAChE,QAAQ,MAAA,gBAAsB;IAC/B,CAAC;WACK,OAAO;AAEd,OAAI,eAAe,MAAA,kBAAyB,OAAA,QAAc,MAAM;AAChE;;AAEF,MAAI,eAAe,MAAA,kBAAyB;EAE5C,MAAM,YAAY,IAAI,IAAI,QAAQ,KAAK,QAAQ,IAAI,OAAO,CAAC;AAC3D,QAAA,MAAY,UAAU,YAAY;GAChC,MAAM,cAAc,IAAI,IACtB,QAAQ,SAAS,MAAO,EAAE,QAAQ,CAAC,EAAE,MAAM,GAAG,EAAE,CAAE,CACnD;GAGD,MAAM,aAAa,QAAQ,QAAQ,MACjC,MAAA,cAAoB,GAAG,WAAW,mBAAmB,CACtD;AAID,UAAO,CACL,GAJoB,QACnB,QAAQ,QAAQ,CAAC,YAAY,IAAI,IAAI,OAAO,CAAC,CAC7C,KAAK,QAAQ,MAAA,QAAc,IAAI,CAAC,EAGjC,GAAG,WACJ;IACD;;;;;;;;CASJ,MAAM,QACJ,UACA,QACA,SACe;AAEf,QAAA,WAAiB;AACjB,QAAA,eAAqB,SAAS;EAC9B,MAAM,sBAAsB,MAAA;EAC5B,MAAM,sBAAsB,MAAA;EAC5B,MAAM,MAAA,GAAA,2BAAA,KAAa;AACnB,QAAA,MAAY,UAAU,YAAY,CAChC,GAAG,SACH;GAAE;GAAI;GAAQ;GAAS,2BAAW,IAAI,MAAM;GAAE,CAC/C,CAAC;AACF,MAAI;GACF,MAAM,MAAM,MAAM,MAAA,KAAW,OAAO,UAAU,MAAA,aAAmB;IAC/D,OAAO;IACP,QAAQa,wBAAAA,iBAAiB,SAAS,QAAQ,SAAS;IACnD,UAAU,SAAS;IACnB,cAAc,SAAS;IACvB,mBAAmB;IACpB,CAAC;AACF,OAAI,MAAA,cAAoB,OAAO,GAAG,EAAE;AAClC,UAAM,MAAA,KAAW,OAAO,UAAU,IAAI,OAAO;AAC7C;;AAEF,SAAA,MAAY,UAAU,YACpB,QAAQ,KAAK,UACX,MAAM,OAAO,KAAK;IAAE,GAAG;IAAO,OAAO,IAAI;IAAQ,GAAG,MACrD,CACF;AAKD,OACE,MAAA,sBAA4B,uBAC5B,MAAA,sBAA4B,oBAEvB,OAAA,eAAqB,SAAS;WAE9B,OAAO;AACd,SAAA,cAAoB,OAAO,GAAG;AAC9B,SAAA,MAAY,UAAU,YACpB,QAAQ,QAAQ,UAAU,MAAM,OAAO,GAAG,CAC3C;AACD,SAAA,QAAc,MAAM;AACpB,SAAM;;;CAIV,MAAM,OAAO,IAA8B;EACzC,MAAM,QAAQ,MAAA,MAAY,aAAa,CAAC,MAAM,MAAM,EAAE,OAAO,GAAG;AAChE,MAAI,CAAC,MAAO,QAAO;AACnB,QAAA,MAAY,UAAU,YAAY,QAAQ,QAAQ,MAAM,EAAE,OAAO,GAAG,CAAC;AACrE,QAAM,MAAA,YAAkB,MAAM;AAC9B,SAAO;;CAGT,MAAM,QAAuB;EAC3B,MAAM,UAAU,MAAA,MAAY,aAAa;AACzC,QAAA,MAAY,eACJI,sBAAAA,YACP;EAMD,MAAM,YAHU,MAAM,QAAQ,WAC5B,QAAQ,KAAK,UAAU,MAAA,YAAkB,MAAM,CAAC,CACjD,EACwB,QACtB,MAAkC,EAAE,WAAW,WACjD;AACD,MAAI,SAAS,WAAW,EACtB,OAAA,QAAc,SAAS,GAAG,OAAO;WACxB,SAAS,SAAS,EAC3B,OAAA,QACE,IAAI,eACF,SAAS,KAAK,MAAM,EAAE,OAAO,EAC7B,GAAG,SAAS,OAAO,MAAM,QAAQ,OAAO,+BACzC,CACF;;CAIL,SAAe;AACb,QAAA;AACA,QAAA,eAAqB;AACrB,QAAA,cAAoB,KAAA;AAEpB,QAAA,WAAiB,KAAA;AACjB,QAAA,gBAAsB,OAAO;AAC7B,QAAA,kBAAwB,IAAI,iBAAiB;AAC7C,eAAa,MAAA,kBAAwB;AACrC,QAAA,oBAA0B,KAAA;;CAG5B,OAAA,YAAmB,OAAuD;AACxE,MAAI,MAAM,SAAS,MAAA,SACjB,OAAM,MAAA,KAAW,OAAO,MAAA,UAAgB,MAAM,MAAM;MAEpD,OAAA,cAAoB,IAAI,MAAM,GAAG;;;;;;;CASrC,gBAAgB,UAAwB;AACtC,MAAI,MAAA,YAAmB;AACvB,QAAA,cAAoB,MAAA,UAAgB,SAAS,CAAC,SAAS,UAAU;AAC/D,OAAI,MAAM,WAAW,YAAa;AAClC,OAAI,MAAM,OAAO,KAAK,UAAU,UAAW;AAC3C,SAAA;AACK,SAAA,eAAqB,SAAS;IACnC;;;;;;;;;CAUJ,OAAA,eAAsB,UAAkB,UAAU,GAAkB;AAClE,MAAI,MAAA,MAAY,aAAa,CAAC,WAAW,EAAG;EAC5C,MAAM,aAAa,MAAA;EACnB,MAAM,qBAAqB,MAAA,aAAmB;AAC9C,MAAI;GACF,MAAM,UAAU,MAAM,MAAA,KAAW,KAAK,UAAU;IAC9C,QAAQ;IACR,OAAO;IACP,QAAQ,MAAA,gBAAsB;IAC/B,CAAC;AACF,OAAI,eAAe,MAAA,kBAAyB;GAC5C,MAAM,kBAAkB,IAAI,IAAI,QAAQ,KAAK,QAAQ,IAAI,OAAO,CAAC;AACjE,SAAA,MAAY,UAAU,YACpB,QAAQ,QAAQ,UACd,MAAA,cAAoB,OAAO,iBAAiB,mBAAmB,CAChE,CACF;WACM,OAAO;AACd,OAAI,eAAe,MAAA,kBAAyB;AAC5C,OAAI,UAAU,GAAG;AACf,UAAA,oBAA0B,iBAAiB;AACzC,WAAA,oBAA0B,KAAA;AACrB,WAAA,eAAqB,UAAU,UAAU,EAAE;OAC/C,IAAK;AACR;;AAEF,SAAA,QAAc,MAAM;;;CAIxB,SAAS,KAAqD;AAC5D,SAAO;GACL,IAAI,IAAI;GACR,OAAO,IAAI;GACX,QAAQ,IAAI,QAAQ;GACpB,WAAW,IAAI,KAAK,IAAI,WAAW;GACpC;;CAGH,eAA4B;AAC1B,SAAO,IAAI,IACT,MAAA,MAAY,aAAa,CAAC,SAAS,MAAO,EAAE,QAAQ,CAAC,EAAE,MAAM,GAAG,EAAE,CAAE,CACrE;;;CAIH,eACE,OACA,iBACA,oBACS;AACT,SACE,CAAC,MAAM,SACP,gBAAgB,IAAI,MAAM,MAAM,IAChC,CAAC,mBAAmB,IAAI,MAAM,MAAM"}