/** * @license * Copyright 2025 Steven Roussey * SPDX-License-Identifier: Apache-2.0 */ import type { IJobExecuteContext } from "@workglow/job-queue"; import { Job } from "@workglow/job-queue"; import type { IExecuteContext, IRunConfig, StreamEvent, TaskConfig, TaskEntitlements } from "@workglow/task-graph"; import { CreateWorkflow, Task } from "@workglow/task-graph"; import type { DataPortSchema, FromSchema } from "@workglow/util/schema"; declare const inputSchema: { readonly type: "object"; readonly properties: { readonly url: { readonly type: "string"; readonly title: "URL"; readonly description: "The URL to fetch from"; readonly format: "uri"; }; readonly method: { readonly enum: readonly ["GET", "HEAD", "POST", "PUT", "DELETE", "PATCH"]; readonly title: "Method"; readonly description: "The HTTP method to use"; readonly default: "GET"; }; readonly headers: { readonly type: "object"; readonly additionalProperties: { readonly type: "string"; }; readonly title: "Headers"; readonly description: "The headers to send with the request"; }; readonly body: { readonly type: "string"; readonly title: "Body"; readonly description: "The body of the request"; }; readonly response_type: { readonly enum: readonly ["stream", "text", "json", "blob", "arraybuffer"]; readonly title: "Response Type"; readonly description: string; }; readonly timeout: { readonly type: "number"; readonly title: "Timeout"; readonly description: "Request timeout in milliseconds"; }; readonly credential_key: { readonly type: "string"; readonly format: "credential"; readonly title: "Credential Key"; readonly description: "Key to look up in the credential store. The resolved secret is placed on the request according to credential_scheme. Incompatible with the queued path, which would persist the secret."; readonly "x-ui-hidden": true; }; readonly credential_scheme: { readonly enum: readonly ["bearer", "basic", "header", "none"]; readonly title: "Credential Scheme"; readonly description: "How the resolved credential is sent. 'bearer' and 'basic' use the Authorization header ('basic' expects an already base64-encoded user:pass); 'header' uses credential_header; 'none' resolves but sends nothing."; readonly default: "bearer"; readonly "x-ui-hidden": true; }; readonly credential_header: { readonly type: "string"; readonly title: "Credential Header"; readonly description: "Header name used when credential_scheme is 'header'. Must be a bare header token (letters, digits, hyphens)."; readonly default: "Authorization"; readonly "x-ui-hidden": true; }; }; readonly required: readonly ["url", "response_type"]; readonly additionalProperties: false; }; declare const outputSchema: { readonly type: "object"; readonly properties: { readonly body: { readonly title: "Body"; readonly description: string; readonly "x-stream": "binary"; readonly format: "binary"; }; readonly json: { readonly title: "JSON"; readonly description: "The JSON response"; }; readonly text: { readonly type: "string"; readonly title: "Text"; readonly description: "The text response"; }; readonly blob: { readonly title: "Blob"; readonly description: "The blob response"; }; readonly arraybuffer: { readonly title: "ArrayBuffer"; readonly description: "The arraybuffer response"; }; readonly metadata: { readonly type: "object"; readonly properties: { readonly contentType: { readonly type: "string"; }; readonly headers: { readonly type: "object"; readonly additionalProperties: { readonly type: "string"; }; }; readonly status: { readonly type: "number"; }; readonly notModified: { readonly type: "boolean"; }; }; readonly required: readonly ["contentType", "headers", "status", "notModified"]; readonly additionalProperties: false; readonly title: "Response Metadata"; readonly description: "HTTP response metadata: content type, headers, status, and 304 state"; }; }; readonly additionalProperties: false; }; export type FetchUrlResponseType = "stream" | "text" | "json" | "blob" | "arraybuffer"; export type FetchUrlTaskInput = FromSchema; export type FetchUrlTaskOutput = FromSchema; /** * Strict `Content-Length` parse, fail-closed. `parseInt` accepts trailing * garbage ("123abc" -> 123), which would let a malformed header defeat the * truncation check this feeds. RFC 9112 §6.3 permits repeated headers to be * combined as "v1, v2": equal duplicates are valid, mismatched values are a * protocol error. Returns `undefined` when the header states no size, which * skips the size assertion: either absent (chunked transfer), or present and * empty — `Headers.get` answers `""` rather than `null` for a proxy that emits * a bare `Content-Length:`, and reading "the origin stated nothing" as a * malformed value would fail such a fetch permanently. */ export declare function parseContentLength(header: string | null, url: string): number | undefined; /** * True when the body arrived under a content coding, so what the runtime hands * back is not what the origin measured. * * Nothing here sets `Accept-Encoding`, and both undici and browsers send * `gzip, deflate, br` on their own and then transparently decode. The origin's * `Content-Length` describes the ENCODED octets while `response.body` yields * the decoded ones, so comparing the two fails every compressed response that * states a length — and drives progress far past 100% on the way. Fetch leaves * `Content-Encoding` on the response after decoding, which is the only signal * that the stated size measures something else. * * `identity` names no coding, so it does not disqualify the length. */ export declare function isContentEncoded(header: string | null): boolean; /** * True when the request carries a validator, which is what makes a `304` a * meaningful answer. An unsolicited `304` is a server protocol error — * "unmodified relative to what?" has no answer a caller can act on — so it * stays an error rather than being reported as `notModified`. */ export declare function hasConditionalHeader(headers: Record | undefined): boolean; export declare class FetchUrlJob extends Job { static readonly type: string; /** * Header names holding a resolved credential, so safeFetch can drop them on a * cross-origin redirect. `Authorization` is covered by safeFetch's own * strip-set; this exists for `credential_scheme: "header"`, whose header name * is caller-chosen and is stripped from the job input before the request is * issued — nothing downstream could otherwise know which header is secret. * * Set by {@link FetchUrlTask} on the inline path only. It is deliberately not * a data port: job inputs are persisted durably by the queued path, and this * names a credential header. The queued path refuses credentials outright, so * there is nothing for it to carry. */ sensitiveHeaders: readonly string[] | undefined; protected issueRequest(input: Input, context: IJobExecuteContext): Promise; /** * Streams the response body as `binary-delta` on the `body` port, then a * `finish` carrying `metadata` plus whichever derived port `response_type` * named. The status is known before the first byte, so a non-2xx throws * before any delta is emitted and no downstream consumer is ever dispatched * on a doomed fetch. * * This both calls `context.emitStreamEvent` AND `yield`s every `body` * delta. A caller must pick exactly one delivery path: a job context whose * `emitStreamEvent` fans out to the same listeners that also drain this * generator's yields would double-ship every chunk. The inline `execute()` * below is safe because it only drains yields (its context carries no * `emitStreamEvent`); a future queued-source caller wiring a worker context * through here needs to route through one path, not both. * * The emitted terminal `finish` is the stream's end-of-stream marker, and * awaiting it is what makes it one — by ordering, not by delivery. On the * awaited fast path the await covers the dispatch to attached listeners; on * the channel path (where that fast path is suppressed and the reassembler's * own dispatch is deliberately unawaited) it covers the durable write, which * fixes the marker's seq after the last delta's. Nothing can overtake it on * either carrier. That is the signal {@link FetchUrlTask.consumeJobStream} * ends on, instead of guessing from the job's completion that no more bytes * are coming. It * carries only `metadata`: the derived port is the *value*, which the * settled job output already carries, and duplicating a whole body into the * carrier's stream log (where a transferable buffer may be detached out from * under the job that still has to return it) buys nothing. */ executeStream(input: Input, context: IJobExecuteContext): AsyncIterable>; /** * Publishes the end-of-stream marker to whatever receiver the worker wired * up, and waits for it to land. A run with no `emitStreamEvent` delivers to * nobody, so there is nothing to mark. */ private emitStreamEnd; execute(input: Input, context: IJobExecuteContext): Promise; } export type FetchUrlTaskConfig = TaskConfig & { queue?: boolean | string; }; /** * Distinct private origins one declaration will name before it gives up and * falls back to the unscoped form. A url array comes from run input, so it can * name as many hosts as the caller likes; past this point the scoped list is no * longer something a human can review, and "every private destination" is the * honest summary of what the run would reach. Staying scoped is a convenience — * widening is always safe, narrowing would not be. */ export declare const MAX_PRIVATE_RESOURCE_PATTERNS = 64; /** * Entitlements a fetch of `url` requires. A task that OWNS a `FetchUrlTask` * must declare these itself: the graph snapshot is taken over * `graph.getTasks()` before any `execute()` runs, so an owned child created * inside `execute()` is never in it. * * `url` may be unknown at evaluation time (root-task input is not applied * yet), in which case this fails closed and requires an unscoped * `network:private` rather than under-declaring it. A known array — the shape * a MapTask projection leaves on the inner fetch — is classified item by * item; any unusable entry fails closed the same way a missing scalar does. */ export declare function fetchUrlEntitlementsFor(url: string | readonly string[] | undefined): TaskEntitlements; export declare class FetchUrlTask extends Task { static type: string; static category: string; static title: string; static description: string; static hasDynamicSchemas: boolean; static hasDynamicEntitlements: boolean; /** * Refuses a subclass that overrides `execute()`. * * {@link executeStream} is the sole implementation, and `TaskRunner` * dispatches a streamable task there — it never calls `execute()`, so an * override of it runs on no path a `run()` takes. That silence is the * hazard: a subclass overriding `execute()` to derive the real URL from a * domain input would have its rewrite skipped and would fetch whatever * unresolved `url` the input happened to carry, with no type error and no * runtime signal. Failing at construction turns an invisible wrong fetch * into an immediate, addressable error. * * {@link resolveFetchInput} is the supported seam for that rewrite, and it * runs on every path. */ constructor(config?: NoInfer>, runConfig?: NoInfer>); /** * Belt and braces with {@link assertMethodAllowsResponseType}: the job layer * fails closed on a persisted payload, and this fails the same combination at * the task layer, before anything is enqueued. */ validateInput(input: Input, skipPorts?: ReadonlySet): Promise; static entitlements(): TaskEntitlements; /** * Dynamic entitlement check: when the configured URL targets a private or * loopback host the task additionally requires `network:private`, scoped * via the URL's origin so grants can be resource-limited (e.g. a dev-mode * grant for `http://localhost:*`). The graph runner evaluates this before * `execute()` runs, so a denied private URL never issues a network call. */ entitlements(): TaskEntitlements; static configSchema(): DataPortSchema; static inputSchema(): { readonly type: "object"; readonly properties: { readonly url: { readonly type: "string"; readonly title: "URL"; readonly description: "The URL to fetch from"; readonly format: "uri"; }; readonly method: { readonly enum: readonly ["GET", "HEAD", "POST", "PUT", "DELETE", "PATCH"]; readonly title: "Method"; readonly description: "The HTTP method to use"; readonly default: "GET"; }; readonly headers: { readonly type: "object"; readonly additionalProperties: { readonly type: "string"; }; readonly title: "Headers"; readonly description: "The headers to send with the request"; }; readonly body: { readonly type: "string"; readonly title: "Body"; readonly description: "The body of the request"; }; readonly response_type: { readonly enum: readonly ["stream", "text", "json", "blob", "arraybuffer"]; readonly title: "Response Type"; readonly description: string; }; readonly timeout: { readonly type: "number"; readonly title: "Timeout"; readonly description: "Request timeout in milliseconds"; }; readonly credential_key: { readonly type: "string"; readonly format: "credential"; readonly title: "Credential Key"; readonly description: "Key to look up in the credential store. The resolved secret is placed on the request according to credential_scheme. Incompatible with the queued path, which would persist the secret."; readonly "x-ui-hidden": true; }; readonly credential_scheme: { readonly enum: readonly ["bearer", "basic", "header", "none"]; readonly title: "Credential Scheme"; readonly description: "How the resolved credential is sent. 'bearer' and 'basic' use the Authorization header ('basic' expects an already base64-encoded user:pass); 'header' uses credential_header; 'none' resolves but sends nothing."; readonly default: "bearer"; readonly "x-ui-hidden": true; }; readonly credential_header: { readonly type: "string"; readonly title: "Credential Header"; readonly description: "Header name used when credential_scheme is 'header'. Must be a bare header token (letters, digits, hyphens)."; readonly default: "Authorization"; readonly "x-ui-hidden": true; }; }; readonly required: readonly ["url", "response_type"]; readonly additionalProperties: false; }; static outputSchema(): { readonly type: "object"; readonly properties: { readonly body: { readonly title: "Body"; readonly description: string; readonly "x-stream": "binary"; readonly format: "binary"; }; readonly json: { readonly title: "JSON"; readonly description: "The JSON response"; }; readonly text: { readonly type: "string"; readonly title: "Text"; readonly description: "The text response"; }; readonly blob: { readonly title: "Blob"; readonly description: "The blob response"; }; readonly arraybuffer: { readonly title: "ArrayBuffer"; readonly description: "The arraybuffer response"; }; readonly metadata: { readonly type: "object"; readonly properties: { readonly contentType: { readonly type: "string"; }; readonly headers: { readonly type: "object"; readonly additionalProperties: { readonly type: "string"; }; }; readonly status: { readonly type: "number"; }; readonly notModified: { readonly type: "boolean"; }; }; readonly required: readonly ["contentType", "headers", "status", "notModified"]; readonly additionalProperties: false; readonly title: "Response Metadata"; readonly description: "HTTP response metadata: content type, headers, status, and 304 state"; }; }; readonly additionalProperties: false; }; /** * Computes output schema dynamically based on the current response_type. * `body` and `metadata` are always present; `response_type` additionally * narrows in the matching derived port (json/text/blob/arraybuffer). */ outputSchema(): DataPortSchema; /** * Collects {@link executeStream} into a single value for callers that want * one. `isTaskStreamable` is true for this task whenever its dynamic output * schema keeps the `x-stream` `body` port, so `TaskRunner` dispatches to * `executeStream` and normally never here — making `executeStream` the sole * implementation and this a thin drain over it. A second implementation * would be dead code that no run or test exercises, so the constructor * refuses a subclass that writes one. * * A `finish` is mandatory: without one there is no output, and returning the * `{}` an absent finish leaves behind would hand the caller an object with * no `metadata` and no derived port, typed as if the fetch had succeeded. */ execute(rawInput: FetchUrlTaskInput, executeContext: IExecuteContext): Promise; /** * Resolves the request this run will actually issue. The default returns * `input` unchanged; a subclass whose input is a domain key (a CIK, an * accession number) rather than a `url` overrides this to build the request * from it. * * Runs first in {@link executeStream} — ahead of the conditional-request * guard, so the guard inspects the headers that will really be sent — and * everything downstream (default queue name, credential refusal, the job * payload handed to the queue) reads the value returned here. * * SECURITY: {@link entitlements} is evaluated against the UNRESOLVED input, * so a rewrite is invisible to it — a resolver returning a private/internal * destination declares no `network:private` and would be granted none. The * resolved destination is therefore re-checked against that declaration (see * {@link assertResolvedDestinationDeclared}) and a rewrite onto an * undeclared private host fails closed. Keep the private origins a subclass * can produce inside the origin its declared input already names, declare * the private input up front, or opt in through * {@link allowsPrivateResolution} when the input names no url at all. */ protected resolveFetchInput(input: FetchUrlTaskInput, context: IExecuteContext): Promise; /** * Whether {@link resolveFetchInput} may return a private/internal * destination when the unresolved input names no url to scope it against. * * Default `false`. A domain-key input (a CIK, an accession number) carries * no url, so {@link entitlements} can only declare an UNSCOPED * `network:private` — a declaration covering every private destination, and * one that is enforced solely when `enforceEntitlements` is set, which it is * not by default. Nothing else states which private host such a resolver is * entitled to, so the default answer is none. * * Returning `true` makes the resolver itself the trust boundary: it may then * reach any private host, and the redirect scope `safeFetch` enforces is * whichever origin the resolver chose. */ protected allowsPrivateResolution(): boolean; /** * Fails closed when {@link resolveFetchInput} rewrote the request onto a * private/internal destination outside what {@link entitlements} declared. * * The declaration is computed from the unresolved `runInputData.url`, while * `FetchUrlJob.issueRequest` classifies the RESOLVED url and passes * `allowPrivate` on the strength of that classification. Left alone, the * rewrite authorizes itself: a task declaring only `network:http` reaches * `http://127.0.0.1/...` with `allowPrivate: true`, and the redirect-scope * enforcement inside `safeFetch` is handed the rewritten origin as its own * scope. The private-network decision has to follow what was declared, so * anything the declaration does not cover is refused before a request is * issued (inline and queued alike — the resolved input is what gets * enqueued). * * A task whose url is unavailable at declaration time declares an unscoped * `network:private` (see {@link entitlements}), so there is no declared scope * here to measure the resolved destination against. That is not a reason to * permit it: the unscoped declaration is fail-closed only where it is * enforced, and `enforceEntitlements` on `IRunConfig` defaults to **false**. * A subclass whose input is a domain key rather than a url — the shape * {@link resolveFetchInput} exists for — would otherwise resolve onto any * private/internal destination and reach it with `allowPrivate: true` on the * default path, which is the same self-authorizing rewrite the declared-url * branch refuses. Such a resolution is therefore refused too, unless the * subclass declares itself the trust boundary via * {@link allowsPrivateResolution}. */ private assertResolvedDestinationDeclared; /** * Runs the fetch, either inline or through a job queue depending on the * `queue` config, and yields the body as `binary-delta` events on `body` * followed by a `finish`. Credential resolution is handled by the input * resolver system — credential_key arrives already resolved to the secret. * * Invariant, scoped to PERSISTENCE: a resolved secret is never written * anywhere durable. Job queues persist their payloads (SQLite/Postgres/SQS), * so the secret is only ever placed on the in-process request headers of the * inline path, and the queued path refuses to run at all when a credential is * present. * * It does leave this method over the wire, which is a different question. The * request carries it to the origin the caller named, and if that origin * redirects, `safeFetch` follows the chain — dropping `Authorization` / * `Cookie` / `Proxy-Authorization` on any hop that crosses origins, plus * whatever {@link FetchUrlJob.sensitiveHeaders} names for the `header` * scheme, whose header is caller-chosen and unrecognizable otherwise. A * stripped header stays stripped for the rest of the chain, so a * vendor -> attacker -> vendor redirect cannot launder it back. */ executeStream(rawInput: FetchUrlTaskInput, executeContext: IExecuteContext): AsyncIterable>; /** * Everything after {@link resolveFetchInput}. Split out so `execute()` can * resolve once and still report the resolved request, rather than resolving * a second time (the seam is a subclass hook and may not be idempotent). */ private streamResolved; /** * Adapts `handle.onStream`'s pushed callbacks into the pulled async iterable * `executeStream` has to be. * * **Where the listener promise actually paces the worker.** It resolves only * once the event it carried has been pulled off this generator, and on the * channel-less fast path that reaches the producer: `JobQueueClient`'s * `handleJobStream` awaits every listener, the worker awaits that dispatch, * and a slow sink therefore parks the job mid-body. When a stream *channel* * subscription is open for the job the fast path is suppressed and the * channel replays the row with its dispatch deliberately unawaited (the * event is already durably published, so on a cross-process carrier there is * no live producer left to pace) — nothing on that path observes this * promise, so the worker runs ahead of the consumer no matter what this * method does. * * Nothing here bounds what a fire-and-forget carrier can hand over: the * events it has already delivered sit in `pending`, and its own log holds * whatever it published. Backpressure across the job boundary exists only * where the dispatch is awaited, and no queue this side keeps can create it. * * **What ends the loop.** `FetchUrlJob` emits a terminal `finish` and awaits * it. That await buys ordering rather than delivery: on the fast path it * covers the dispatch to attached listeners, and on the channel path the * durable write, which fixes the marker's seq after the last delta's. Either * way nothing can overtake it, so a marker arriving here means every delta * ahead of it already did — the body is whole. The loop ends there. The job's * settled output remains the authority for the *value*, so the marker is * released without being re-yielded: `waitFor()` produces the one `finish`, * and the two can never disagree or arrive twice. * * Settlement is the fallback for a stream that produces no marker — a failed * job (`waitFor()` rejects and there is nothing more to wait for), or a * carrier that dropped the terminal row. Because the completion signal and * the stream are independent transports, ending there immediately would drop * an event still in flight, which on a body port is silent truncation. So a * settled stream keeps taking turns of the event loop for as long as each * turn actually drains something, and ends on the first turn that does not. * A residual remains for that fallback alone: an event landing more than one * idle turn after the last, on a carrier that does not deliver the marker * *within that turn*, is still lost. Delivering both a moment after the * grace turn is spent does not save it — the loop is already gone, so the * marker has nothing left to end. * * An `error` event is not decoration on any of this: it is the only in-band * report of a failure the completion signal may not carry, so it is queued * in order and raised as a throw when the consumer reaches it. * * **`requireStreamDelivery` closes the gap a handle's own capability cannot.** * `onStream` is advertised whenever the client has an attached server, which * on a durable queue is the *possibility* of delivery rather than a * guarantee: the job may be claimed by a worker in another process, whose * events never reach here. Which process claims a job is not knowable at * `send` time, so the only honest advertise-time answer would be to withhold * `onStream` from every durable queue — refusing streaming for the ordinary * single-process deployment that works today. The discriminator is available * here instead, and it is the terminal marker rather than a delta count: a * locally-claimed job always emits it (`FetchUrlJob.executeStream` calls * `emitStreamEnd` whenever the context carries `emitStreamEvent`, which * `JobQueueWorker.executeJob` supplies unconditionally), even for a 304 or a * zero-length body, while a remotely-claimed one produces nothing and this * loop exits through its settlement fallback with `streamEnded` false. */ private consumeJobStream; /** * Whether this run will store its output anywhere — the question the * conditional-request guard turns on, since a stored bodiless `304` is what * destroys the artifact it validated. * * Two things have to hold for a row to be written, and both are checked. * * The task has to be cacheable at all. Every write path in `CacheCoordinator` * — the row save, and the stream sinks that mint a `CacheRef` for the `body` * port — returns early on `task.cacheable`, so a `cacheable: false` instance * (or a subclass declaring `static cacheable = false`) can overwrite nothing * and the refusal would cost a caller a working conditional request for a * hazard that cannot occur. * * And a cache has to be in play. The run's own resolution is the authority * (see {@link IExecuteContext.cacheRegistry}): reading `runConfig.outputCache` * alone answered for one of the three ways a cache reaches a run and left the * other two — the config passed to `run()`, and a `CACHE_REGISTRY` binding — * silently unguarded. Reading it FIRST was wrong in the other direction: * `run(input, { outputCache: false })` resolves no cache, and the instance * field it overrides would still have refused the run. A context that went * through a runner always carries the `cacheRegistry` key, `undefined` * included, so the resolution answers whenever there is one; the legacy field * speaks only for a hand-built context, which has no resolution to consult. */ private hasOutputCache; private prepareJobInput; private resolveOrCreateQueue; /** * Detects when response_type changes and emits schemaChange so consumers * see the dynamic output schema update. */ setInput(input: Partial): void; private getDefaultQueueName; } export declare const fetchUrl: (input: FetchUrlTaskInput, config?: FetchUrlTaskConfig) => Promise; declare module "@workglow/task-graph" { interface Workflow { fetch: CreateWorkflow; } } export {};