feat(tasks): workers started with versions only claim their own tasks

OpenWorkflow 0.10.1 claims any run regardless of workflow name and fails
unregistered ones with 5s-5m backoff; filter the claim SQL by name@version.

Co-Authored-By: Claude Opus 5.5 <[email protected]>

packages/tasks/README.md

@@ -41,6 +41,8 @@
 
 - 不兼容的改动(改步骤名或顺序、改输入结构、改已有步骤的语义)必须把 `version` +1,同时**保留旧版本定义并注册**。worker 用 `startWorker({ versions: ["submitDownload@1", "submitDownload@2"] })` 或者省略参数(表示全部版本)。等 `list({ taskName, status })` 查不到旧版本的 run 了,再删除旧定义。
 - 新 run 一律用最高版本。旧版本 run 如果找不到对应的 worker,会无限退避等待,不会失败。
+- 带 `versions` 的 worker 只认领这些 (任务名, 版本) 的 run(本包在认领 SQL 上加了过滤;OpenWorkflow 0.10.1 原生认领不看任务名,
+  认领到未注册的 run 会以"未注册"失败并按 5s→5m 退避,分组 worker 会互相拖慢)。server 用它把 worker 分成 io 组与 agent 组。
 - 同一版本里,只允许在末尾追加步骤,或插入幂等步骤。
 
 ## 已知缺口(OpenWorkflow 0.10.1)
@@ -55,4 +57,4 @@
 
 ## 测试
 
-`deno task test`(或本包 `deno task test`)会连接开发库,每个测试 schema 先用 `@umt/db` 的 `migrateToLatest` 建全部业务表(`DATABASE_URL`,默认 `postgres://umt:umt@localhost:55433/umt`)。每个测试文件使用独立的 `tasks_test_<名>_<pid+随机>` schema(并发跑多份也不冲突),测试前后都会清理;库不可达时跳过集成测试。覆盖范围:kill -9 后续跑(子进程 worker)、取消、幂等 enqueue、两个 relay 并发、事实先到达、fencing。
+`deno task test`(或本包 `deno task test`)会连接开发库,每个测试 schema 先用 `@umt/db` 的 `migrateToLatest` 建全部业务表(`DATABASE_URL`,默认 `postgres://umt:umt@localhost:55433/umt`)。每个测试文件使用独立的 `tasks_test_<名>_<pid+随机>` schema(并发跑多份也不冲突),测试前后都会清理;库不可达时跳过集成测试。覆盖范围:kill -9 后续跑(子进程 worker)、取消、幂等 enqueue、两个 relay 并发、事实先到达、fencing、分组 worker 只认领自己的任务。

packages/tasks/src/openworkflow-engine.ts

@@ -59,6 +59,54 @@ export async function migrateEngineSchema(databaseUrl: string, schema = ENGINE_S
   await backend.stop();
 }
 
+/**
+ * 包一层只认领指定 (任务名, 版本) 的 backend:与 BackendPostgres.claimWorkflowRun 同一条 SQL(截止过期处理 +
+ * SKIP LOCKED 认领),candidate 多一个 workflow_name/version 过滤。其余方法经原型链委托给原 backend。
+ */
+function claimFilteredBackend(backend: BackendPostgres, defs: readonly TaskDefinition[]): BackendPostgres {
+  // biome-ignore lint/suspicious/noExplicitAny: 访问 BackendPostgres 的私有字段(pg / namespaceId / workflowRunsTable)
+  const b = backend as any;
+  const keys = [...new Set(defs.map((d) => `${d.name}@${engineVersion(d.version)}`))];
+  const filtered = Object.create(backend) as BackendPostgres;
+  Object.defineProperty(filtered, "claimWorkflowRun", {
+    value: async (params: { workerId: string; leaseDurationMs: number }) => {
+      const table = b.workflowRunsTable();
+      const [claimed] = await b.pg`
+      WITH expired AS (
+        UPDATE ${table}
+        SET "status" = 'failed', "error" = ${b.pg.json({ message: "Workflow run deadline exceeded" })},
+            "worker_id" = NULL, "available_at" = NULL, "finished_at" = NOW(), "updated_at" = NOW()
+        WHERE "namespace_id" = ${b.namespaceId}
+          AND "status" IN ('pending', 'running', 'sleeping')
+          AND "deadline_at" IS NOT NULL
+          AND "deadline_at" <= NOW()
+        RETURNING "id"
+      ),
+      candidate AS (
+        SELECT "id"
+        FROM ${table}
+        WHERE "namespace_id" = ${b.namespaceId}
+          AND "status" IN ('pending', 'running', 'sleeping')
+          AND "available_at" <= NOW()
+          AND ("deadline_at" IS NULL OR "deadline_at" > NOW())
+          AND ("workflow_name" || '@' || COALESCE("version", '')) = ANY(${keys}::text[])
+        ORDER BY CASE WHEN "status" = 'pending' THEN 0 ELSE 1 END, "available_at", "created_at"
+        LIMIT 1
+        FOR UPDATE SKIP LOCKED
+      )
+      UPDATE ${table} AS wr
+      SET "status" = 'running', "attempts" = "attempts" + 1, "worker_id" = ${params.workerId},
+          "available_at" = NOW() + ${params.leaseDurationMs} * INTERVAL '1 millisecond',
+          "started_at" = COALESCE("started_at", NOW()), "updated_at" = NOW()
+      FROM candidate
+      WHERE wr."id" = candidate."id" AND wr."namespace_id" = ${b.namespaceId}
+      RETURNING wr.*;`;
+      return claimed ?? null;
+    },
+  });
+  return filtered;
+}
+
 export class OpenWorkflowEngine implements TaskEngine {
   private readonly ow: OpenWorkflow;
   private readonly workers: Worker[] = [];
@@ -144,7 +192,10 @@ export class OpenWorkflowEngine implements TaskEngine {
   async startWorker(opts: StartWorkerOptions = {}) {
     const defs = this.opts.registry.select(opts.versions);
     const workflows = defs.map((d) => this.toWorkflow(d)) as Workflow<unknown, unknown, unknown>[];
-    const worker = new Worker({ backend: this.backend, workflows, concurrency: opts.concurrency ?? 4 });
+    // 指定了 versions 的 worker 只认领这些任务:OpenWorkflow 0.10.1 认领不看任务名,认领到未注册的 run 会
+    // 以"未注册"失败并指数退避(5s→5m),分组 worker(io / agent)会互相拖慢对方的任务。
+    const backend = opts.versions?.length ? claimFilteredBackend(this.backend, defs) : this.backend;
+    const worker = new Worker({ backend, workflows, concurrency: opts.concurrency ?? 4 });
     await worker.start();
     this.workers.push(worker);
     this.log.info("worker started", { tasks: defs.map((d) => `${d.name}@${d.version}`) });

packages/tasks/test/worker-groups.test.ts

@@ -0,0 +1,80 @@
+import { afterAll, beforeAll, describe, expect, it } from "vitest";
+import { z } from "zod";
+import { defineTask, type OpenWorkflowEngine, TaskRegistry } from "../src/index.ts";
+import {
+  connectDb,
+  dropSchemas,
+  migrateBizSchema,
+  pgAvailable,
+  resetSchemas,
+  setupEngine,
+  type TestDb,
+  testSchemas,
+  waitFor,
+} from "./helpers.ts";
+
+// worker 分组(io / agent):带 versions 的 worker 只认领自己的任务,不会误认领后以"未注册"失败再退避。
+
+const schemas = testSchemas("groups");
+const hasPg = await pgAvailable();
+
+/** 两个"进程"各一套注册表(任务集相同,handler 记下自己是哪一组) */
+function registryFor(group: string, log: string[]) {
+  const mk = (name: string) =>
+    defineTask({
+      name,
+      version: 1,
+      inputSchema: z.object({ i: z.number() }),
+      handler: async (ctx) => {
+        log.push(`${group}:${name}`);
+        return { group, i: ctx.input.i };
+      },
+    });
+  return new TaskRegistry().register(mk("ioTask"), mk("runAgent"), mk("resolveDecision"));
+}
+
+describe.skipIf(!hasPg)("worker 分组认领", () => {
+  let db: TestDb;
+  let io: OpenWorkflowEngine;
+  let agent: OpenWorkflowEngine;
+  const log: string[] = [];
+
+  beforeAll(async () => {
+    await resetSchemas(schemas);
+    db = connectDb(schemas.biz);
+    await migrateBizSchema(db, schemas.biz);
+    io = await setupEngine(db, schemas, registryFor("io", log));
+    agent = await setupEngine(db, schemas, registryFor("agent", log));
+    await io.startWorker({ versions: ["ioTask"], concurrency: 4 });
+    await agent.startWorker({ versions: ["runAgent", "resolveDecision"], concurrency: 1 });
+  }, 30_000);
+
+  afterAll(async () => {
+    await io?.stop();
+    await agent?.stop();
+    await db?.destroy();
+    await dropSchemas(schemas);
+  }, 60_000);
+
+  it("io 组只跑 ioTask,agent 组只跑 runAgent/resolveDecision;每个 run 一次认领即完成", async () => {
+    const runs: { runId: string; task: string }[] = [];
+    for (let i = 0; i < 6; i++) {
+      for (const task of ["ioTask", "runAgent", "resolveDecision"]) {
+        const r = await io.enqueue(task, { i }, { idempotencyKey: `${task}-${i}`, traceId: `t-${i}` });
+        runs.push({ runId: r.runId, task });
+      }
+    }
+    for (const r of runs) {
+      const v = await waitFor(async () => {
+        const x = await io.get(r.runId);
+        return x?.status === "completed" ? x : null;
+      }, 20_000);
+      expect(v.output).toMatchObject({ group: r.task === "ioTask" ? "io" : "agent" });
+      // 误认领会以 "not registered" 失败并退避 → attempts > 1
+      expect(v.attempts).toBe(1);
+    }
+    expect(log.filter((l) => l.startsWith("io:")).every((l) => l === "io:ioTask")).toBe(true);
+    expect(log.filter((l) => l.startsWith("agent:")).every((l) => l !== "agent:ioTask")).toBe(true);
+    expect(log).toHaveLength(18);
+  }, 60_000);
+});