require("../_virtual/_rolldown/runtime.cjs");
let _langchain_core_utils_uuid = require("@langchain/core/utils/uuid");
//#region src/stream/submit-coordinator.ts
/**
* Owns the run-submission lifecycle for a single
* {@link StreamController}.
*
* # What this module is
*
* The {@link SubmitCoordinator} is the piece of the controller that
* dispatches runs (`submit()`), enforces multitask strategies, queues
* deferred submissions, races dispatch against terminal lifecycle
* events, and surfaces errors back through the per-submit `onError`
* callback and the root snapshot.
*
* Conceptually a submit looks like:
*
*   1. Optionally rebind to a different thread (`options.threadId`).
*   2. Mint a thread id if one isn't bound yet.
*   3. Wait for the controller's root pump to be ready (so the
*      transport is subscribed before the run is dispatched —
*      otherwise we could miss replayed events).
*   4. Apply the {@link StreamSubmitOptions.multitaskStrategy} to
*      decide whether to abort, enqueue, reject, or proceed.
*   5. Race the dispatch promise (`thread.submitRun()`) against the next root
*      terminal lifecycle event.
*   6. Settle the resulting state (loading flag, error slot) and
*      drain the next queued submission, if any.
*
* # Why it lives in its own class
*
* The submit lifecycle is the most state-heavy part of the
* controller — six promises, an abort controller, a queue, a
* terminal-vs-command race, and bidirectional callback wiring with
* the controller. Splitting it out keeps `controller.ts` focused on
* subscription / projection wiring while letting the submit logic
* evolve independently.
*
* # Why we race "command" against "terminal"
*
* For fast runs, the server's terminal lifecycle event can arrive
* *before* the dispatch HTTP response has resolved. Racing the two
* lets us detect terminal early and not block waiting for a now-stale
* dispatch response. The dispatch response is still consumed (via
* `.then(notifyCreated).catch(reportError)`) so `onCreated` still
* fires and dispatch errors still surface through `onError`.
*
* # Queue semantics (`multitaskStrategy: "enqueue"`)
*
* When a run is already in flight, an `"enqueue"` submit is recorded
* into {@link queueStore} and the call returns immediately. After the
* active run terminates, `#drainQueue` schedules the head of the
* queue as a fresh submit on the next macrotask. Each drained
* submission has its own `multitaskStrategy` cleared so it doesn't
* recursively re-enqueue.
*
* @see StreamController - The owner; injects every collaborator dep.
*/
function terminalReason(event) {
	if (event === "completed") return "success";
	if (event === "failed") return "error";
	if (event === "interrupted") return "interrupt";
	return "stopped";
}
/**
* Frozen empty queue value used as the initial / cleared snapshot.
*
* Reusing one frozen reference keeps store identity stable across
* empty resets, so React's `useSyncExternalStore` doesn't think the
* queue changed when it actually didn't.
*/
const EMPTY_QUEUE = Object.freeze([]);
/**
* Coordinates one controller's run-submission lifecycle.
*
* The constructor takes a bag of callbacks rather than a reference to
* the parent {@link StreamController} on purpose:
*
*   - It keeps the dependency surface explicit and testable — every
*     piece of controller state the submit lifecycle touches is one
*     of these closures.
*   - It avoids a cyclic dependency between controller and coordinator.
*   - Tests can construct one with stub callbacks and assert behavior
*     without mocking the entire controller.
*
* @typeParam StateType         - Root state shape.
* @typeParam InterruptType     - Root interrupt payload shape.
* @typeParam ConfigurableType  - `config.configurable` shape accepted
*   by submit (usually `Record<string, unknown>`).
*/
var SubmitCoordinator = class {
	/** Controller-level options forwarded into `submitRun` / callbacks. */
	#options;
	/** Root snapshot store; written for `isLoading`, `error`, `interrupts`. */
	#rootStore;
	/** Pending submissions awaiting the active run to terminate. */
	#queueStore;
	/** Probes the controller's `disposed` flag from deferred work. */
	#getDisposed;
	/** Reads the controller's currently-bound thread id. */
	#getCurrentThreadId;
	/** Updates the controller's thread id (used when minting a new id). */
	#setCurrentThreadId;
	/** Records a thread id we created client-side so hydrate can skip a 404 round-trip. */
	#rememberSelfCreatedThreadId;
	/** Drops a thread id from the self-created set once it's committed server-side. */
	#forgetSelfCreatedThreadId;
	/** Triggers a hydrate on the controller (used by `options.threadId` rebinds). */
	#hydrate;
	/** Lazily creates / returns the active {@link ThreadStream}. */
	#ensureThread;
	/** Starts the previously-deferred root pump after a self-created thread commits. */
	#startDeferredRootPump;
	/** Abandons a deferred root pump after a self-created dispatch fails. */
	#abandonDeferredRootPump;
	/** Resolves once the controller's root subscription pump is up. */
	#waitForRootPumpReady;
	/** Resolves on the next root terminal lifecycle (or on abort). */
	#awaitNextTerminal;
	/**
	* Resolves on the resumed run's terminal, skipping stale `interrupted`
	* events from the run being resumed (see {@link dispatchResume}).
	*/
	#awaitResumedRunTerminal;
	/** Called once at the start of every {@link submit} invocation. */
	#onSubmitStart;
	/** Marks that a local run dispatch is now active. */
	#onRunStart;
	/** Records a server-accepted local run id and fires `onCreated`. */
	#onRunCreated;
	/** Fires `onCompleted` for the local run lifecycle. */
	#onRunCompleted;
	/** Marks the local run dispatch lifecycle as settled. */
	#onRunEnd;
	/**
	* Apply a submit input optimistically before dispatch. Returns the
	* id-injected payload to dispatch plus a handle for terminal
	* reconciliation, or `undefined` when optimistic UI is disabled / no
	* echo applies (dispatch the raw input).
	*/
	#beginOptimistic;
	/** Reconcile optimistic state when a run terminates. */
	#settleOptimistic;
	/**
	* Active submission's abort controller. `undefined` between submits.
	*
	* Used both for `multitaskStrategy: "rollback"` (abort the previous
	* controller's signal) and `stop()` (abort the current one without
	* starting a new one).
	*/
	#runAbort;
	constructor(params) {
		this.#options = params.options;
		this.#rootStore = params.rootStore;
		this.#queueStore = params.queueStore;
		this.#getDisposed = params.getDisposed;
		this.#getCurrentThreadId = params.getCurrentThreadId;
		this.#setCurrentThreadId = params.setCurrentThreadId;
		this.#rememberSelfCreatedThreadId = params.rememberSelfCreatedThreadId;
		this.#forgetSelfCreatedThreadId = params.forgetSelfCreatedThreadId;
		this.#hydrate = params.hydrate;
		this.#ensureThread = params.ensureThread;
		this.#startDeferredRootPump = params.startDeferredRootPump;
		this.#abandonDeferredRootPump = params.abandonDeferredRootPump;
		this.#waitForRootPumpReady = params.waitForRootPumpReady;
		this.#awaitNextTerminal = params.awaitNextTerminal;
		this.#awaitResumedRunTerminal = params.awaitResumedRunTerminal;
		this.#onSubmitStart = params.onSubmitStart ?? (() => void 0);
		this.#onRunStart = params.onRunStart ?? (() => void 0);
		this.#onRunCreated = params.onRunCreated ?? (() => void 0);
		this.#onRunCompleted = params.onRunCompleted ?? (() => void 0);
		this.#onRunEnd = params.onRunEnd ?? (() => void 0);
		this.#beginOptimistic = params.beginOptimistic ?? (() => void 0);
		this.#settleOptimistic = params.settleOptimistic ?? (() => void 0);
	}
	/**
	* Submit input to the active thread.
	*
	* Honours {@link StreamSubmitOptions.multitaskStrategy}:
	*
	*   - `"rollback"` (default) — aborts any in-flight run and
	*     dispatches immediately.
	*   - `"reject"`              — throws synchronously when a run is
	*     already in flight.
	*   - `"enqueue"`             — defers via {@link #enqueueSubmission};
	*     the call returns without dispatching.
	*   - `"interrupt"`           — falls through to the default path
	*
	* Errors are routed through both the per-submit `onError` callback
	* and `rootStore.error`. Aborts (controller dispose / rollback) are
	* silently dropped.
	*
	* To resume a pending interrupt, use {@link StreamController.respond}
	* instead of `submit()`.
	*
	* @param input   - Input payload for the run.
	* @param options - Per-submit options (config, metadata, callbacks,
	*   strategy, etc).
	*/
	async submit(input, options) {
		if (this.#getDisposed()) return;
		this.#onSubmitStart();
		const overrideThreadId = options?.threadId;
		if (overrideThreadId !== void 0 && overrideThreadId !== this.#getCurrentThreadId()) await this.#hydrate(overrideThreadId);
		const wasSelfCreated = this.#getCurrentThreadId() == null;
		if (wasSelfCreated) {
			const threadId = (0, _langchain_core_utils_uuid.v7)();
			this.#setCurrentThreadId(threadId);
			this.#rememberSelfCreatedThreadId(threadId);
			this.#options.onThreadId?.(threadId);
			this.#rootStore.setState((s) => ({
				...s,
				threadId
			}));
		}
		const currentThreadId = this.#getCurrentThreadId();
		if (currentThreadId == null) return;
		const thread = this.#ensureThread(currentThreadId, wasSelfCreated);
		const activeThreadId = currentThreadId;
		const strategy = options?.multitaskStrategy ?? "rollback";
		const hasActiveRun = !wasSelfCreated && this.#runAbort != null && !this.#runAbort.signal.aborted;
		if (hasActiveRun && strategy === "reject") throw new Error("submit() rejected: a run is already in flight and multitaskStrategy is 'reject'.");
		if (hasActiveRun && strategy === "enqueue") {
			this.#enqueueSubmission(input, options);
			return;
		}
		this.#runAbort?.abort();
		const abort = new AbortController();
		this.#runAbort = abort;
		this.#rootStore.setState((s) => ({
			...s,
			interrupts: [],
			interrupt: void 0,
			error: void 0,
			isLoading: true
		}));
		let optimisticHandle;
		let dispatchInput = input;
		let createdRunId;
		let pendingCompletionReason;
		let completionNotified = false;
		let settleEvent;
		const notifyCompletion = (reason) => {
			if (completionNotified) return;
			if (createdRunId == null) {
				pendingCompletionReason = reason;
				return;
			}
			completionNotified = true;
			this.#onRunCompleted(reason, createdRunId);
		};
		const reportError = (error) => {
			if (abort.signal.aborted) return;
			this.#rootStore.setState((s) => ({
				...s,
				error
			}));
			try {
				options?.onError?.(error);
			} catch {}
		};
		try {
			const prepared = this.#beginOptimistic(input);
			if (prepared != null) {
				optimisticHandle = prepared.handle;
				dispatchInput = prepared.dispatchInput;
			}
			await this.#waitForRootPumpReady();
			const boundConfig = bindThreadConfig(options?.config, currentThreadId);
			const terminalPromise = this.#awaitNextTerminal(abort.signal);
			this.#onRunStart();
			let terminalSettled = false;
			let terminal;
			const commandPromise = thread.submitRun({
				input: dispatchInput ?? null,
				config: boundConfig,
				metadata: options?.metadata ?? void 0,
				forkFrom: options?.forkFrom,
				multitaskStrategy: options?.multitaskStrategy === "enqueue" ? "enqueue" : options?.multitaskStrategy
			});
			commandPromise.then(() => {
				this.#startDeferredRootPump();
				this.#forgetSelfCreatedThreadId(activeThreadId);
			}, () => {
				if (wasSelfCreated) {
					this.#abandonDeferredRootPump();
					this.#forgetSelfCreatedThreadId(activeThreadId);
				}
			});
			const notifyCreated = (result) => {
				if (typeof result.run_id !== "string") return;
				createdRunId = result.run_id;
				this.#onRunCreated(createdRunId);
				if (pendingCompletionReason != null) notifyCompletion(pendingCompletionReason);
			};
			const first = await Promise.race([terminalPromise.then((value) => ({
				type: "terminal",
				value
			})), commandPromise.then((result) => ({
				type: "command",
				result
			}), (error) => ({
				type: "error",
				error
			}))]);
			if (first.type === "error") throw first.error;
			if (first.type === "command") notifyCreated(first.result);
			else {
				terminal = first.value;
				terminalSettled = true;
				commandPromise.then(notifyCreated).catch((error) => {
					if (!terminalSettled) reportError(error);
				});
			}
			terminal ??= await terminalPromise;
			terminalSettled = true;
			settleEvent = terminal.event;
			if (terminal.event === "failed" && !abort.signal.aborted) {
				const runError = new Error(terminal.error ?? "Run failed with no error message");
				this.#rootStore.setState((s) => ({
					...s,
					error: runError
				}));
				try {
					options?.onError?.(runError);
				} catch {}
			}
			notifyCompletion(terminalReason(terminal.event));
		} catch (error) {
			if (!abort.signal.aborted) settleEvent = "failed";
			reportError(error);
		} finally {
			this.#rootStore.setState((s) => ({
				...s,
				isLoading: false
			}));
			if (this.#runAbort === abort) this.#runAbort = void 0;
			if (optimisticHandle != null) this.#settleOptimistic(optimisticHandle, abort.signal.aborted ? "aborted" : settleEvent ?? "failed");
			this.#onRunEnd();
			setTimeout(() => this.#drainQueue(), 0);
		}
	}
	/**
	* Surface a *resumed* run's failure the same way {@link submit} surfaces
	* a fresh run's failure — by writing it to the reactive
	* {@link RootSnapshot.error} slot.
	*
	* `respond()` / `respondAll()` dispatch their `input.respond` command on
	* the controller directly (they target a specific interrupt, so they
	* cannot go through {@link submit}, which only does `run.start`). The
	* resumed run therefore never passed through the submit lifecycle that
	* populates `rootStore.error` — only the persistent lifecycle listener
	* observed it, and that listener drives `isLoading` alone. Without this,
	* a resumed run that fails (e.g. a missing model key surfaced after the
	* user approves an interrupt) would flip `isLoading` back to `false`
	* with `error` left untouched, so `stream.error`-driven UIs (error
	* banners, API-key retry prompts) would silently miss it.
	*
	* The `dispatch` thunk is awaited, so a dispatch failure rejects the
	* caller's `respond()` *and* lands in `rootStore.error`. The resumed
	* run's terminal is watched in the **background** so the returned promise
	* still settles on dispatch — preserving the resume command's
	* resolve-on-dispatch contract (and avoiding a hang when no terminal is
	* ever emitted, e.g. in unit tests).
	*
	* Reuses the shared {@link #runAbort} slot, so `stop()`, `dispose()`, and
	* a rollback `submit()` all cancel the terminal watch (no spurious error
	* on user-initiated cancel) and treat the resumed run as the active run.
	*
	* The terminal watch uses {@link #awaitResumedRunTerminal}, which skips
	* stale `interrupted` terminals from the run being resumed (they can reach
	* the pump after `input.requested` but before `respondInput` calls
	* `#prepareForNextRun`) and only accepts a later `interrupted` once a
	* root `running` lifecycle for the resumed run has been observed.
	*
	* @param dispatch - Sends the `input.respond` command (and marks the
	*   targeted interrupt resolved). Invoked after the terminal watch is
	*   armed.
	* @param optimisticHandle - Optional handle from an optimistic `update`
	*   applied before dispatch (HITL "push card into state + resume"). Settled
	*   on the resumed run's terminal — pending messages flip to `sent` and
	*   un-echoed non-message keys roll back on failure — exactly like the
	*   `submit()` optimistic lifecycle. A dispatch failure settles it `failed`.
	*/
	async dispatchResume(dispatch, optimisticHandle) {
		if (this.#getDisposed()) return;
		this.#runAbort?.abort();
		const abort = new AbortController();
		this.#runAbort = abort;
		this.#rootStore.setState((s) => s.error === void 0 ? s : {
			...s,
			error: void 0
		});
		const reportError = (error) => {
			if (abort.signal.aborted) return;
			this.#rootStore.setState((s) => ({
				...s,
				error
			}));
		};
		let optimisticSettled = false;
		const settleOptimisticOnce = (event) => {
			if (optimisticSettled || optimisticHandle == null) return;
			optimisticSettled = true;
			this.#settleOptimistic(optimisticHandle, event);
		};
		this.#awaitResumedRunTerminal(abort.signal).then((terminal) => {
			if (this.#runAbort === abort) this.#runAbort = void 0;
			if (terminal.event === "failed" && !abort.signal.aborted) reportError(new Error(terminal.error ?? "Run failed with no error message"));
			settleOptimisticOnce(abort.signal.aborted ? "aborted" : terminal.event);
			setTimeout(() => this.#drainQueue(), 0);
		});
		try {
			await dispatch();
		} catch (error) {
			reportError(error);
			settleOptimisticOnce("failed");
			if (this.#runAbort === abort) this.#runAbort = void 0;
			throw error;
		}
	}
	/**
	* Abort the current run (if any) and force `isLoading=false`.
	*
	* Client-side only — server-side cancel is handled by
	* {@link StreamController.stop} before this is invoked.
	*/
	async stop() {
		this.abortActiveRun();
		this.#rootStore.setState((s) => ({
			...s,
			isLoading: false
		}));
	}
	/**
	* Abort the current run without forcing the loading flag down.
	*
	* Used by {@link StreamController.dispose}: disposal already tears
	* down the root store, so flipping `isLoading` here is unnecessary
	* and would race the dispose path.
	*/
	abortActiveRun() {
		this.#runAbort?.abort();
		this.#runAbort = void 0;
	}
	/**
	* Cancel a queued submission by id.
	*
	* @param id - Client-side queue entry id to remove.
	* @returns `true` when the entry was found and dropped, `false` otherwise.
	*/
	async cancelQueued(id) {
		const current = this.#queueStore.getSnapshot();
		const next = current.filter((entry) => entry.id !== id);
		if (next.length === current.length) return false;
		this.#queueStore.setState(() => next);
		return true;
	}
	/**
	* Drop every queued submission. Server-side cancel arrives with A0.3.
	*/
	async clearQueue() {
		this.#queueStore.setState(() => EMPTY_QUEUE);
	}
	/**
	* Append a submission to the queue without dispatching.
	*
	* The drained submission is later run via {@link #drainQueue} after
	* the active run terminates.
	*/
	#enqueueSubmission(input, options) {
		const entry = {
			id: (0, _langchain_core_utils_uuid.v7)(),
			values: input ?? void 0,
			options,
			createdAt: /* @__PURE__ */ new Date()
		};
		this.#queueStore.setState((current) => [...current, entry]);
	}
	/**
	* Drain the head of the queue if no run is active.
	*
	* Called from the `finally` block of `submit()` on the next
	* macrotask (so the just-finished run's state flushes first).
	* Strips the strategy off the dequeued options to prevent infinite
	* re-enqueueing.
	*/
	#drainQueue() {
		if (this.#getDisposed()) return;
		if (this.#runAbort != null && !this.#runAbort.signal.aborted) return;
		const current = this.#queueStore.getSnapshot();
		if (current.length === 0) return;
		const [next, ...rest] = current;
		this.#queueStore.setState(() => rest);
		const nextOptions = {
			...next.options ?? {},
			multitaskStrategy: void 0
		};
		this.submit(next.values, nextOptions).catch(() => {});
	}
};
/**
* Merge `thread_id` into a user-supplied `config.configurable` blob.
*
* The platform expects `config.configurable.thread_id` on every run
* dispatch; we set it last so user-supplied values can't accidentally
* override the active thread id (which would route the run to a
* different thread).
*/
function bindThreadConfig(config, threadId) {
	const base = config != null && typeof config === "object" ? config : {};
	const configurable = base.configurable != null && typeof base.configurable === "object" ? base.configurable : {};
	return {
		...base,
		configurable: {
			...configurable,
			thread_id: threadId
		}
	};
}
//#endregion
exports.EMPTY_QUEUE = EMPTY_QUEUE;
exports.SubmitCoordinator = SubmitCoordinator;

//# sourceMappingURL=submit-coordinator.cjs.map