Notify tasks: merge main, real-transport tests, Slack structural broadcasts, webhook redaction

packages/job-queue/src/job/JobQueueWorker.ts

@@ -1109,7 +1109,17 @@ export class JobQueueWorker<
     try {
       job.status = JobStatus.PENDING;
       const nextAvailableTime = await this.limiter.getNextAvailableTime();
-      job.visibleAt = retryDate instanceof Date ? retryDate : nextAvailableTime;
+      // `instanceof Date` alone is a type check wearing a validity check's
+      // clothes: an Invalid Date passes it, its NaN reaches `delaySeconds`
+      // below, and `claim.retry` throws a RangeError out of `toISOString()` —
+      // caught here, so the job is silently never rescheduled at all. The
+      // date comes from a job, provider or third-party error object, any of
+      // which can construct one. No clamping happens here: whatever parses
+      // the remote hint owns the retry-delay policy; this worker only refuses
+      // the impossible.
+      const usableRetryDate =
+        retryDate instanceof Date && Number.isFinite(retryDate.getTime()) ? retryDate : undefined;
+      job.visibleAt = usableRetryDate ?? nextAvailableTime;
       job.progress = 0;
       job.progressMessage = "";
       job.progressDetails = null;

packages/job-queue/src/job/__tests__/JobQueueWorker.test.ts

@@ -6,6 +6,7 @@
 
 import type { IJobExecuteContext, JobStorageFormat } from "@workglow/job-queue";
 import {
+  DelayLimiter,
   InMemoryQueueStorage,
   InMemoryRateLimiterStorage,
   Job,
@@ -429,3 +430,69 @@
     await storage.deleteAll();
   });
 });
+
+describe("JobQueueWorker retry date validity", () => {
+  // An Invalid Date passes `instanceof Date`, so it used to be accepted as the
+  // job's next visible time. Its NaN then reached `claim.retry`'s
+  // `delaySeconds`, where `new Date(NaN).toISOString()` throws a RangeError —
+  // swallowed by rescheduleJob's own catch, leaving the claim dropped and the
+  // job never rescheduled at all. The observable bug is a stranded job, not a
+  // bad timestamp.
+  it("falls back to the limiter time when handed an invalid retry date", async () => {
+    const queueName = `retry-date-${uuid4()}`;
+    const storage = new InMemoryQueueStorage<TI, TO>(queueName);
+    await storage.migrate();
+    const { messageQueue, jobStore } = wrapQueueStorage(storage);
+
+    // A limiter time distinct from "now" so the fallback is unmistakable.
+    const limiter = new DelayLimiter();
+    const fallbackTime = new Date(Date.now() + 60_000);
+    await limiter.setNextAvailableTime(fallbackTime);
+
+    const worker = new JobQueueWorker<TI, TO, TJob>(TJob, {
+      messageQueue,
+      jobStore,
+      queueName,
+      pollIntervalMs: 5,
+      stopTimeoutMs: 0,
+      limiter,
+    });
+
+    const id = await storage.add({
+      input: { taskType: "default", data: "x" },
+      visible_at: null,
+      completed_at: null,
+      deadline_at: null,
+    } as any);
+
+    const claims = await messageQueue.receive({ workerId: "test-worker", leaseMs: 30_000, max: 1 });
+    expect(claims.length).toBe(1);
+    const claim = claims[0]!;
+
+    const retryDelays: (number | undefined)[] = [];
+    const originalRetry = claim.retry.bind(claim);
+    (claim as { retry: (opts?: { delaySeconds?: number }) => Promise<void> }).retry = async (
+      opts
+    ) => {
+      retryDelays.push(opts?.delaySeconds);
+      await originalRetry(opts);
+    };
+
+    const job = new TJob({ queueName, input: { taskType: "default", data: "x" }, id: claim.id });
+    // @ts-expect-error reaching the private claim registry the worker settles through
+    worker.activeClaims.set(claim.id, claim);
+
+    // @ts-expect-error rescheduleJob is protected; the invalid date is the point
+    await worker.rescheduleJob(job, new Date(NaN));
+
+    expect(retryDelays.length).toBe(1);
+    expect(Number.isFinite(retryDelays[0]!)).toBe(true);
+    expect(job.visibleAt.getTime()).toBe(fallbackTime.getTime());
+
+    const row = await storage.get(claim.id as any);
+    expect(row?.status).toBe(JobStatus.PENDING);
+    expect(Number.isFinite(Date.parse(String(row?.visible_at)))).toBe(true);
+
+    await storage.deleteAll();
+  });
+});

packages/tasks/README.md

@@ -9,6 +9,9 @@ A package of task types for common operations, workflow management, and data pro
 - [Quick Start](#quick-start)
 - [Available Tasks](#available-tasks)
   - [FetchUrlTask](#fetchurltask)
+  - [WebhookNotifyTask](#webhooknotifytask)
+  - [SlackNotifyTask](#slacknotifytask)
+  - [DiscordNotifyTask](#discordnotifytask)
   - [DebugLogTask](#debuglogtask)
   - [DelayTask](#delaytask)
   - [JavaScriptTask](#javascripttask)
@@ -30,7 +33,7 @@ bun add @workglow/tasks
 ## Quick Start
 
 ```typescript
-import { Workflow, fetch, debugLog, delay } from "@workglow/tasks";
+import { Workflow } from "@workglow/tasks";
 
 // Simple workflow example (fluent API)
 const workflow = new Workflow()
@@ -42,19 +45,27 @@ const results = await workflow.run();
 ```
 
 ```typescript
-import { FetchUrlTask, DebugLogTask, DelayTask } from "@workglow/tasks";
+import { fetchUrl, debugLog, delay } from "@workglow/tasks";
 
-// Simple sequence using Task classes directly
-const fetchResult = await new FetchUrlTask({
+// Simple sequence using the exported helpers
+const fetchResult = await fetchUrl({
   url: "https://api.example.com/data",
   response_type: "json",
-}).run();
+});
 
-await new DebugLogTask({ console: fetchResult.json }, { log_level: "info" }).run();
+await debugLog({ console: fetchResult.json }, { log_level: "info" });
 
-await new DelayTask({}, { delay: 1000 }).run();
+await delay({}, { delay: 1000 });
 ```
 
+> **Inputs go to `run()`, not to the constructor.** A task constructor takes
+> `(config, runConfig)`, and `TaskConfigSchema` is `additionalProperties: false`,
+> so `new SomeTask({ url: "…" })` throws a `TaskConfigurationError` before any
+> work happens. Each helper above (`fetchUrl`, `debugLog`, `slackNotify`, …) is
+> just `new SomeTask(config).run(input)` with the two arguments in the right
+> places. To bake input values into an instance, put them under the config's
+> `defaults` key — see the [WebhookNotifyTask](#webhooknotifytask) examples.
+
 ```typescript
 import { fetch, debugLog, delay } from "@workglow/tasks";
 
@@ -92,14 +103,14 @@ Makes HTTP requests with built-in retry logic, progress tracking, and multiple r
 
 ```typescript
 // Simple GET request
-const response = await new FetchUrlTask({
+const response = await fetchUrl({
   url: "https://api.example.com/users",
   response_type: "json",
-}).run();
+});
 console.log(response.json);
 
 // POST request with headers
-const postResponse = await new FetchUrlTask({
+const postResponse = await fetchUrl({
   url: "https://api.example.com/users",
   method: "POST",
   headers: {
@@ -108,13 +119,13 @@ const postResponse = await new FetchUrlTask({
   },
   body: JSON.stringify({ name: "John", email: "john@example.com" }),
   response_type: "json",
-}).run();
+});
 
 // Text response
-const textResponse = await new FetchUrlTask({
+const textResponse = await fetchUrl({
   url: "https://example.com/readme.txt",
   response_type: "text",
-}).run();
+});
 console.log(textResponse.text);
 ```
 
@@ -126,6 +137,181 @@ console.log(textResponse.text);
 - Queue-based rate limiting (requires creation of a `@workglow/job-queue` instance)
 - Comprehensive error handling
 
+### WebhookNotifyTask
+
+Sends a JSON payload to a webhook endpoint via HTTP POST.
+
+A webhook URL is treated as a secret throughout all three notification tasks: for
+Slack and Discord the token is part of the URL path, so the URL is kept out of the
+output schema and error messages report only the endpoint's origin.
+
+**Input Schema:**
+
+- `url` (string, optional): Webhook endpoint to POST to. Kept out of errors and output, but a value set here is stored verbatim in the graph JSON — use `url_credential_key` to keep the secret out of the saved workflow.
+- `payload` (object, required): JSON body to send
+- `headers` (object, optional): Additional headers, merged over the JSON content type
+- `timeout` (number, optional): Request timeout in milliseconds. Default: `30000`
+- `allow_private_destination` (boolean, optional): Permit posting to a private/internal/loopback destination. Requires the `network:private` entitlement. Default: `false`
+- `url_credential_key` (string, optional): Credential store key whose resolved value is the entire webhook URL — the secret itself, not a bearer token. Takes precedence over `url`. A value that is not an absolute `http(s)` URL (e.g. a bearer token) is rejected with a configuration error.
+
+**Output Schema:**
+
+- `success` (boolean): Always `true`; a non-2xx response throws
+- `status` (number): HTTP status code returned by the endpoint
+- `response` (string): Response body, truncated to 1KB. Always empty for a private/internal destination
+
+**Examples:**
+
+```typescript
+// Direct task usage
+const result = await webhookNotify({
+  url: "https://example.com/hooks/abc123",
+  payload: { event: "deploy", version: "1.4.2" },
+  headers: { "X-Signature": "sha256=..." },
+});
+console.log(result.status);
+
+// Config vs. input: the constructor takes CONFIG, so a fixed endpoint belongs
+// under `defaults` — the per-run payload is still passed to `run()`.
+const notifier = new WebhookNotifyTask({
+  title: "Deploy hook",
+  defaults: { url: "https://example.com/hooks/abc123" },
+});
+await notifier.run({ payload: { event: "deploy", version: "1.4.2" } });
+
+// In a workflow
+const workflow = new Workflow()
+  .fetch({ url: "https://api.example.com/build" })
+  .webhookNotify({ url: "https://example.com/hooks/abc123", payload: { event: "build" } });
+```
+
+**Features:**
+
+- Runs inline through the SSRF-aware `safeFetch` wrapper
+- **Redirects are refused** — a webhook that answers `3xx` fails rather than re-sending the payload and headers to the new origin. Point `url` at the final endpoint
+- **A private/internal destination is refused unless `allow_private_destination` is set** — which also declares the `network:private` entitlement, scoped to `url` when no credential key is configured. Without the flag the post fails with `PRIVATE_DENIED` before any request is made, and, when an entitlement enforcer is registered, the `network:private` grant is re-checked at execute time against the URL actually resolved — so a run-input or credential-supplied private destination cannot slip past the declaration the enforcer graded
+- A permitted private/internal destination is reachable but its response body is **never echoed** — `response` is always `""`. Notification needs no reply body, and returning one would make this task an SSRF read primitive (e.g. POSTing to a cloud metadata endpoint and reading the answer back into the graph)
+- 429/503 and 5xx raise `RetryableJobError`; retries require a `@workglow/job-queue` consumer, which these inline tasks do not have
+- Response bodies are read as a stream and abandoned past 1MB, so an endpoint answering with an unbounded body cannot exhaust runner memory — on the failure path too
+- Requests time out after 30s by default (`timeout`); a caller abort surfaces as an abort error rather than a retryable network failure
+- A configured `url_credential_key` upgrades the `credential` entitlement from optional to enforced
+- Never cached — the task is side-effecting (`cachePolicy: { kind: "none" }`)
+
+### SlackNotifyTask
+
+Sends a message to a Slack incoming webhook.
+
+**Input Schema:**
+
+- `url` (string, optional): Slack incoming webhook URL. Kept out of errors and output, but a value set here is stored verbatim in the graph JSON — use `url_credential_key` to keep the secret out of the saved workflow.
+- `text` (string, required): Message text, also used as the notification fallback for block messages
+- `blocks` (array, optional): Slack Block Kit blocks
+- `username` (string, optional): Overrides the display name of the posting bot
+- `icon_emoji` (string, optional): Overrides the bot icon, e.g. `:rocket:`
+- `allow_mentions` (boolean, optional): Send `text` unmodified. Default: `false`
+- `timeout` (number, optional): Request timeout in milliseconds. Default: `30000`
+- `allow_private_destination` (boolean, optional): Permit posting to a private/internal/loopback destination. Requires the `network:private` entitlement. Default: `false`
+- `url_credential_key` (string, optional): Credential store key whose resolved value is the entire webhook URL — the secret itself, not a bearer token. Takes precedence over `url`.
+
+**Output Schema:**
+
+- `success` (boolean): Always `true`; a non-2xx response throws
+- `status` (number): HTTP status code returned by Slack
+
+**Examples:**
+
+```typescript
+// Plain message
+await slackNotify({
+  url: "https://hooks.slack.com/services/T000/B000/xxx",
+  text: "Deploy finished",
+});
+
+// Block Kit message with a bot identity
+await slackNotify({
+  url: "https://hooks.slack.com/services/T000/B000/xxx",
+  text: "Deploy finished",
+  blocks: [{ type: "section", text: { type: "mrkdwn", text: "*Deploy finished*" } }],
+  username: "deploybot",
+  icon_emoji: ":rocket:",
+});
+
+// In a workflow
+const workflow = new Workflow().slackNotify({
+  url: "https://hooks.slack.com/services/T000/B000/xxx",
+  text: "Pipeline complete",
+});
+```
+
+**Features:**
+
+- Absent optional fields are omitted from the payload rather than sent as `null`
+- **Redirects are refused** — a webhook that answers `3xx` fails rather than re-sending the payload and headers to the new origin. Point `url` at the final endpoint
+- **A private/internal destination is refused unless `allow_private_destination` is set**, which also declares the `network:private` entitlement, and, when an entitlement enforcer is registered, the `network:private` grant is re-checked at execute time against the URL actually resolved — so a run-input or credential-supplied private destination cannot slip past the declaration the enforcer graded
+- Slack answers `200` with the body `ok`; failure bodies (`invalid_payload`, `no_service`) are surfaced in the error message — but **only for a public destination**. A private/internal endpoint's reply body is never spliced into the error, which would otherwise make the task an SSRF read primitive; its status is still reported
+- **Channel-wide broadcasts in `text` and `blocks` are neutralized by default**, by two mechanisms, because Slack has two ways to say the same thing. _Lexical:_ Slack has no `allowed_mentions`; its documented control is HTML-entity escaping, so the literal `<!` is escaped to `&lt;!`. That defuses `<!channel>`, `<!here>`, `<!everyone>` and `<!subteam^ID>` while leaving `<https://…|label>` links and single-user `<@U123>` mentions intact, and `link_names: false` is sent explicitly. `blocks` is walked to every string leaf, so a broadcast written inside a section, a `fields[]` entry or an `elements[]` entry is defused too. _Structural:_ a `rich_text` message expresses the same ping as an element shape with no `<!` in it at all — `{type: "broadcast", range: "channel"}` or `{type: "usergroup", usergroup_id: "S…"}` — so such an element is rewritten to a plain text node (`@channel`) rather than deleted, since dropping it could leave an `elements[]` empty and Slack rejects that. Residual: a new structural notification element type Slack introduces later is uncovered until it is added to the recognized set. Set `allow_mentions: true` to send both verbatim
+- Requests time out after 30s by default (`timeout`)
+- Response bodies are capped at 1MB while being read
+- Webhook token never appears in error messages, `error.url`, `error.stack`, or task output
+
+### DiscordNotifyTask
+
+Sends a message to a Discord webhook.
+
+**Input Schema:**
+
+- `url` (string, optional): Discord webhook URL. Kept out of errors and output, but a value set here is stored verbatim in the graph JSON — use `url_credential_key` to keep the secret out of the saved workflow.
+- `content` (string, required): Message content
+- `username` (string, optional): Overrides the display name of the webhook
+- `avatar_url` (string, optional): Overrides the avatar of the webhook
+- `embeds` (array, optional): Discord embed objects
+- `allow_mentions` (boolean, optional): Let the message ping. Default: `false`
+- `timeout` (number, optional): Request timeout in milliseconds. Default: `30000`
+- `allow_private_destination` (boolean, optional): Permit posting to a private/internal/loopback destination. Requires the `network:private` entitlement. Default: `false`
+- `url_credential_key` (string, optional): Credential store key whose resolved value is the entire webhook URL — the secret itself, not a bearer token. Takes precedence over `url`.
+
+**Output Schema:**
+
+- `success` (boolean): Always `true`; a non-2xx response throws
+- `status` (number): HTTP status code returned by Discord, `204` on success
+
+**Examples:**
+
+```typescript
+// Plain message
+await discordNotify({
+  url: "https://discord.com/api/webhooks/123/xxx",
+  content: "Build passed",
+});
+
+// Embed with a custom identity
+await discordNotify({
+  url: "https://discord.com/api/webhooks/123/xxx",
+  content: "Build passed",
+  username: "ci",
+  avatar_url: "https://example.com/ci.png",
+  embeds: [{ title: "workglow", description: "All checks green" }],
+});
+
+// In a workflow
+const workflow = new Workflow().discordNotify({
+  url: "https://discord.com/api/webhooks/123/xxx",
+  content: "Pipeline complete",
+});
+```
+
+**Features:**
+
+- A successful post answers `204 No Content`, so no response body is read or parsed
+- **Redirects are refused** — a webhook that answers `3xx` fails rather than re-sending the payload and headers to the new origin. Point `url` at the final endpoint
+- **A private/internal destination is refused unless `allow_private_destination` is set**, which also declares the `network:private` entitlement, and, when an entitlement enforcer is registered, the `network:private` grant is re-checked at execute time against the URL actually resolved — so a run-input or credential-supplied private destination cannot slip past the declaration the enforcer graded
+- A failure body is surfaced in the error message **only for a public destination**; a private/internal endpoint's reply body is never spliced in, which would otherwise make the task an SSRF read primitive
+- Rate limits arrive as `429` and may carry the delay as `{"retry_after": <seconds>}` in the body instead of a `Retry-After` header; both are parsed onto the raised `RetryableJobError`. Nothing acts on the value — retries require a `@workglow/job-queue` consumer, which these inline tasks do not have
+- **Mass mentions are suppressed by default** — `allowed_mentions: { parse: [] }` is sent, so `@everyone`/`@here`, role and user pings in `content` do nothing even when the content was piped in from a fetch or a model. Set `allow_mentions: true` to let the message ping
+- Requests time out after 30s by default (`timeout`)
+- Response bodies are capped at 1MB while being read
+- Webhook token never appears in error messages, `error.url`, `error.stack`, or task output
+
 ### DebugLogTask
 
 Provides console logging functionality with multiple log levels for debugging task graphs.