Files

T

phptits b1f17cf637 v2 platform foundation (docs 01/02/03): Action Runtime + Registry + unified API

- v2 foundation: ActionResult/DevError/ActorContext/ResourceRef/Execution types
- ActionRegistry + ResourceRegistry (namespaced types, FAIL FAST on duplicates)
- ActionRuntime pipeline: validate -> authorize -> middleware -> handler ->
  output validation -> audit -> event -> execution persisted (devos_action_executions)
- Legacy bridge: v1 services registered as devops.*/core.* resources & actions (47 actions, 9 resources)
- Unified Action API: POST /api/v2/actions/:actionId, GET discovery (actions/resources)
- 10 v2 unit tests (registry fail-fast, validation, authz, failure audit, events)

2026-09-10 21:09:44 +00:00

49 KiB

Raw Blame History

09_TASK_JOB_QUEUE_RUNTIME.md

DevOS Developer Work OS

Task / Job / Queue / Worker / Scheduler Runtime 设计规范

Version: 1.0

1. 文档目标

本文定义 DevOS 的统一异步执行基础设施：

Task
Job
Queue
Worker
Execution
Scheduler
Retry
Progress
Cancellation
Concurrency

核心目标：

Agent、Plugin、Workflow、API、UI、CLI 都只能创建 Task / 调用 Action，不允许各自实现自己的任务队列和 Worker。

DevOS 最终应该形成唯一执行链：

User / Agent / Plugin / Workflow
            ↓
           Task
            ↓
           Job
            ↓
          Queue
            ↓
          Worker
            ↓
          Action
            ↓
     Action Runtime
            ↓
        Execution
            ↓
          Event

2. 为什么必须统一 Task Runtime

例如用户：

给我生成 100 篇文章。

错误实现：

Agent
 ↓
for 1..100
 ↓
LLM
 ↓
API
 ↓
LLM
 ↓
API

这样会导致：

Agent 长时间占用
Token 浪费
无法恢复
无法暂停
无法并发控制
无法查看单项进度
一个失败导致整个流程混乱
Agent 崩溃后任务消失

正确实现：

Agent
 ↓
Task
   ├── Job 001
   ├── Job 002
   ├── Job 003
   ├── ...
   └── Job 100
        ↓
      Queue
        ↓
      Workers

Agent 只负责：

创建任务
观察结果
调整策略

不负责：

Queue
Worker
Retry Loop
Concurrency

3. Task / Job / Execution 的区别

必须严格区分：

Task
= 用户/Agent 要完成的一项工作

Job
= Task 中可独立调度的一次工作单元

Execution
= Action 实际执行的一次运行记录

例如：

Task:
生成 100 篇 SEO 文章

Jobs:
article-001
article-002
...
article-100

Execution:
article-001.generate
article-002.generate
...

4. 核心边界

Agent
  ↓
决定做什么
  ↓
Task
  ↓
决定有哪些 Job
  ↓
Queue
  ↓
决定什么时候执行
  ↓
Worker
  ↓
执行 Job
  ↓
Action Runtime
  ↓
产生 Execution

Golden Rule：

Task 描述工作，Job 描述调度单位，Execution 描述实际执行。

5. Task

Task 是用户可理解的业务工作单元。

例如：

SEO Audit
Generate Articles
Deploy Application
Sync GitHub
Backup Database
Import Products

定义：

interface Task {
  id: string;

  type: string;

  name: string;

  description?: string;

  status: TaskStatus;

  actor: ActorRef;

  organizationId: string;

  projectId?: string;

  input?: unknown;

  output?: unknown;

  progress: TaskProgress;

  priority: TaskPriority;

  createdAt: Date;

  startedAt?: Date;

  completedAt?: Date;

  metadata?: Record<string, unknown>;
}

6. Task Status

统一状态：

type TaskStatus =
  | "pending"
  | "queued"
  | "running"
  | "paused"
  | "waiting"
  | "completed"
  | "failed"
  | "cancelled"
  | "partially_completed";

状态：

pending
   ↓
queued
   ↓
running
   ↓
completed

失败：

running
   ↓
failed

暂停：

running
   ↓
paused
   ↓
queued

取消：

pending / queued / running
        ↓
     cancelled

7. Task 不等于 Job

例如：

Task:
部署 myapp

可以包含：

Job 1: Build Docker Image
Job 2: Push Image
Job 3: Connect Server
Job 4: Deploy Container
Job 5: Health Check

因此：

1 Task
+
N Jobs

8. Job

Job 是 Queue 可以调度的最小工作单元。

interface Job {
  id: string;

  taskId: string;

  type: string;

  actionId: string;

  input: unknown;

  status: JobStatus;

  priority: number;

  attempts: number;

  maxAttempts: number;

  scheduledAt?: Date;

  startedAt?: Date;

  completedAt?: Date;

  workerId?: string;

  executionId?: string;

  dependencies?: string[];

  metadata?: Record<string, unknown>;
}

9. Job Status

type JobStatus =
  | "pending"
  | "queued"
  | "leased"
  | "running"
  | "retrying"
  | "completed"
  | "failed"
  | "cancelled";

完整流程：

pending
 ↓
queued
 ↓
leased
 ↓
running
 ↓
completed

失败：

running
 ↓
retrying
 ↓
queued

最终：

failed

10. Execution

Execution 是 Action Runtime 的实际执行记录。

例如：

Job:
github.repository.sync

Execution:
exec_123456

定义：

interface Execution {
  id: string;

  jobId?: string;

  actionId: string;

  actor: ActorRef;

  status: ExecutionStatus;

  input?: unknown;

  output?: unknown;

  error?: ExecutionError;

  startedAt?: Date;

  completedAt?: Date;

  durationMs?: number;

  metadata?: Record<string, unknown>;
}

11. 三者关系

Task
 │
 ├── Job
 │    └── Execution
 │
 ├── Job
 │    └── Execution
 │
 └── Job
      └── Execution

一个 Job 在重试时：

Job
 ├── Execution #1 → failed
 ├── Execution #2 → failed
 └── Execution #3 → completed

因此：

Job 是逻辑工作，Execution 是尝试。

12. 为什么 Retry 不能创建新 Task

错误：

Task A
 ↓
failed
 ↓
Task B

正确：

Task A
 ↓
Job A
 ├── Execution 1 failed
 ├── Execution 2 failed
 └── Execution 3 completed

这样整个任务的历史保持完整。

13. Queue

Queue 是 Job 的调度层。

interface JobQueue {
  enqueue(job: Job): Promise<void>;

  dequeue(queueName: string): Promise<Job | null>;

  acknowledge(jobId: string): Promise<void>;

  retry(jobId: string, delayMs: number): Promise<void>;

  cancel(jobId: string): Promise<void>;
}

14. Queue 不执行业务逻辑

Queue 只负责：

保存
排序
等待
分发
重试调度

不能：

Queue
 ↓
GitHub API

必须：

Queue
 ↓
Worker
 ↓
Action Runtime

15. Queue 类型

推荐至少：

default
high
low
ai
network
browser
deployment
sync

但不要过度创建 Queue。

可以通过：

priority
concurrency
resource group

实现细粒度调度。

16. Queue Priority

定义：

type TaskPriority =
  | "critical"
  | "high"
  | "normal"
  | "low";

内部转换：

critical → 100
high     → 75
normal   → 50
low      → 25

Worker 优先执行高优先级 Job。

17. Worker

Worker 是真正执行 Job 的进程。

interface Worker {
  id: string;

  name: string;

  queues: string[];

  status: WorkerStatus;

  concurrency: number;

  capabilities?: string[];

  startedAt: Date;

  lastHeartbeatAt?: Date;
}

18. Worker Status

starting
idle
busy
draining
stopped
unhealthy

19. Worker 不应该拥有业务状态

Worker 不应该保存：

Task state
User state
Agent state
Plugin state

Worker 只负责：

receive Job
execute
report

这样 Worker 可以随时：

restart
scale
replace

20. Worker 执行流程

Worker
 ↓
Receive Job
 ↓
Acquire Lease
 ↓
Validate Job
 ↓
Create Execution
 ↓
Action Runtime
 ↓
Report Progress
 ↓
Complete Execution
 ↓
Update Job
 ↓
Update Task
 ↓
Publish Event

21. Lease

Worker 获取 Job 时必须建立 Lease。

例如：

Job:
abc

Lease:
worker-03

Expires:
22:31:00

防止：

Worker A
 ↓
Job

Worker A crash

Worker B
 ↓
same Job

导致重复执行。

22. Lease Timeout

如果 Worker 心跳消失：

Lease expired

Job：

leased

重新变成：

queued

然后其他 Worker 可以领取。

23. Heartbeat

Worker：

每 5 秒

发送：

heartbeat

例如：

interface WorkerHeartbeat {
  workerId: string;

  timestamp: Date;

  activeJobs: number;

  capacity: number;
}

24. Worker Graceful Shutdown

Worker 停止时：

SIGTERM
 ↓
draining
 ↓
停止接收新 Job
 ↓
等待当前 Job
 ↓
完成
 ↓
stopped

如果超时：

release leases

让其他 Worker 接管。

25. Worker Concurrency

例如：

Worker A
concurrency = 5

最多：

5 Jobs

同时执行。

不同 Worker：

Worker A → 5
Worker B → 5
Worker C → 10

总并发：

26. Resource Concurrency

除了 Worker，还需要限制外部资源。

例如：

OpenAI
maxConcurrency = 10

GitHub
maxConcurrency = 20

Browser
maxConcurrency = 5

否则：

100 Jobs
 ↓
100 requests
 ↓
API rate limit

27. Resource Group

Job 可以声明：

resourceGroup?: string;

例如：

ai.openai
github.api
browser.instance
server.production

Scheduler / Queue Runtime 可以保证：

同一 resourceGroup
最多 N 个并发

28. Concurrency Policy

interface ConcurrencyPolicy {
  key: string;

  maxConcurrent: number;

  scope:
    | "global"
    | "organization"
    | "project"
    | "resource";
}

例如：

OpenAI
Global:
100

Organization:
10

Project:
5

29. Task Progress

Task 必须支持统一进度：

interface TaskProgress {
  total?: number;

  completed: number;

  failed: number;

  running: number;

  percentage?: number;

  message?: string;

  updatedAt: Date;
}

例如：

SEO Article Generation

72 / 100
72%

30. Job Progress

Job 也可以报告：

0%
25%
50%
75%
100%

例如：

Download website

Worker：

25%
50%
75%
100%

Task Runtime 汇总。

31. Progress Event

统一事件：

task.progress
job.progress

例如：

{
  type: "job.progress",
  jobId: "job_123",
  percentage: 75
}

前端通过：

WebSocket
SSE
Polling

获取。

32. Cancellation

Task 必须支持：

Cancel

但：

Cancel
≠
kill process

流程：

User
 ↓
Cancel Task
 ↓
Task Runtime
 ↓
Signal Job
 ↓
AbortSignal
 ↓
Action Runtime
 ↓
Plugin

33. AbortSignal

ActionContext 已经定义：

signal: AbortSignal;

Worker 必须传递：

Task
 ↓
Job
 ↓
Execution
 ↓
ActionContext.signal

Plugin / Action 必须响应取消。

34. Cancellation State

推荐：

pending → cancelled
queued → cancelled
running → cancelling → cancelled

增加：

"cancelling"

可以避免误认为 Action 已立即停止。

35. 强制终止

如果 Action 长时间不响应：

Grace Period
 ↓
Force Termination

但只能由：

Worker Runtime

执行。

不能由 UI 直接杀 Worker。

36. Retry

Job 可以定义：

interface RetryPolicy {
  maxAttempts: number;

  backoff:
    | "fixed"
    | "linear"
    | "exponential";

  initialDelayMs: number;

  maxDelayMs?: number;

  jitter?: boolean;
}

37. Retry Example

Attempt 1
 ↓
fail
 ↓
1 sec
 ↓
Attempt 2
 ↓
fail
 ↓
2 sec
 ↓
Attempt 3
 ↓
fail
 ↓
4 sec
 ↓
Attempt 4

38. Retryable Error

Action Error 必须声明：

interface ExecutionError {
  code: string;

  message: string;

  retryable: boolean;

  category?: string;

  details?: unknown;
}

例如：

HTTP 429
retryable = true

而：

INVALID_INPUT
retryable = false

39. Retry 分类

推荐：

network
timeout
rate_limit
temporary
dependency
authentication
authorization
validation
business
system

默认策略：

network → retry
timeout → retry
rate_limit → retry
temporary → retry

validation → no retry
authorization → no retry
invalid_input → no retry

40. Dead Letter Queue

最终失败的 Job：

Retry exhausted
 ↓
Dead Letter Queue

例如：

DLQ
├── job-123
├── job-456
└── job-789

管理员可以：

Retry
Inspect
Cancel
Replay

41. Replay

Replay 必须生成新的 Execution。

不能修改旧 Execution。

例如：

Job
 ├── Execution 1 failed
 ├── Execution 2 failed
 └── Execution 3 ← replay

Audit 保留完整历史。

42. Idempotency

所有可能重试的 Action 必须考虑幂等。

例如：

createRepository

第一次：

GitHub 创建成功

但 Worker 在收到响应前崩溃。

系统认为：

failed

再次执行可能产生：

duplicate repository

所以必须支持：

idempotencyKey

43. Job Idempotency

interface Job {
  idempotencyKey?: string;
}

Action Runtime：

same key
 ↓
same logical operation

如果已经成功：

return existing result

44. Task Idempotency

用户重复点击：

Generate 100 articles

不能创建：

Task A
Task B
Task C

除非用户明确要求。

可以使用：

taskType
organization
project
idempotencyKey

进行去重。

45. Scheduling

Scheduler 负责：

什么时候创建 Task / Job

例如：

Every 10 minutes
Daily
Weekly
At specific time
After event

Scheduler 不负责执行。

46. Scheduler Architecture

Scheduler
 ↓
Detect Trigger
 ↓
Create Task / Job
 ↓
Queue
 ↓
Worker

47. Cron

支持：

0 * * * *
0 0 * * *
0 0 * * 1

例如：

每小时同步 GitHub

Scheduler：

cron
 ↓
github.repository.sync Task

48. Schedule Resource

Schedule 本身也是 Resource：

scheduler.schedule

Action：

scheduler.schedule.create
scheduler.schedule.update
scheduler.schedule.delete
scheduler.schedule.enable
scheduler.schedule.disable

这样 UI / API / Agent 都统一。

49. Schedule Definition

interface Schedule {
  id: string;

  name: string;

  cron: string;

  timezone: string;

  actionId?: string;

  taskType?: string;

  input?: unknown;

  enabled: boolean;

  organizationId: string;

  projectId?: string;

  nextRunAt?: Date;

  lastRunAt?: Date;
}

50. Timezone

Schedule 必须明确：

timezone

不能默认：

server timezone

例如：

Asia/Tokyo
UTC
America/New_York

51. Missed Schedule

服务器停机：

00:00 schedule

但服务器：

23:00 - 02:00
offline

重启后怎么办必须由策略决定：

skip
run_once
run_all

默认：

skip

避免突然创建大量任务。

52. Event Trigger

Task 也可以由 Event 创建。

例如：

github.pull_request.created

触发：

code review Task

流程：

Event
 ↓
Trigger
 ↓
Task
 ↓
Job
 ↓
Queue

53. Webhook Trigger

例如：

GitHub Webhook
 ↓
DevOS Event
 ↓
Task

Webhook Controller 只负责：

receive
verify
publish event

不负责业务执行。

54. Task Dependencies

Task 可以依赖其他 Task：

Build
 ↓
Deploy
 ↓
Health Check

Job 也可以依赖 Job：

Job A
 ↓
Job B
 ↓
Job C

55. Job Dependency Graph

定义：

interface JobDependency {
  jobId: string;

  dependsOn: string[];

  condition?:
    | "all"
    | "any"
    | "success"
    | "completed";
}

56. DAG

推荐使用 DAG：

        Build
       /     \
     Test    Security Scan
       \     /
        Deploy
           ↓
       Health Check

不能出现：

A → B
B → C
C → A

必须检测循环。

57. Parallel Jobs

例如：

Build
 ↓
├── Test
├── Security Scan
└── Lint

三者可以并发。

完成后：

Test
Security
Lint
 ↓
Deploy

58. Task Aggregation

Task 状态由 Jobs 聚合：

100 jobs

如果：

90 completed
5 failed
5 running

Task：

partially_completed

进度：

90%

59. Task Completion Policy

支持：

all_success
best_effort
any_success
threshold

例如：

100 articles

要求：

至少 95 篇成功

则：

threshold = 95%

60. Batch Task

针对大量工作：

Task
 ↓
Batch
 ↓
Jobs

例如：

生成 1000 篇文章

不要：

1000 个 Agent Run

应该：

1 Agent Goal
 ↓
1 Task
 ↓
1000 Jobs

61. Batch Job Generation

Agent 可以告诉 Task Runtime：

createBatchTask({
  type: "content.generate",
  items: articles
});

Task Runtime：

items
 ↓
Job Factory
 ↓
Jobs

Agent 不负责逐个创建。

62. Job Factory

interface JobFactory {
  createJobs(
    task: Task,
    input: unknown
  ): Promise<Job[]>;
}

例如：

100 URLs
 ↓
100 Jobs

63. Dynamic Job Creation

有些 Task 不知道最终 Job 数量。

例如：

爬取网站

流程：

Job 1:
Discover URLs

 ↓

发现 500 URLs

 ↓

创建 500 Jobs

允许 Worker / Action 发布：

Task Expansion

但必须通过：

Task Runtime

完成。

64. Dynamic Expansion 限制

必须设置：

maxJobs
maxDepth
maxExpansion

防止：

Job
 ↓
100 jobs
 ↓
10000 jobs
 ↓
1,000,000 jobs

65. Backpressure

当 Queue：

100000 jobs

不能继续无限创建。

Task Runtime 应返回：

accepted
queued
throttled

例如：

Queue capacity exceeded

进入：

waiting

66. Rate Limiting

Job 可以受：

Global Rate Limit
Organization Rate Limit
Project Rate Limit
Plugin Rate Limit
External API Rate Limit

共同约束。

67. Token Budget

AI Job 可以声明：

interface JobBudget {
  maxTokens?: number;

  maxCost?: number;

  maxDurationMs?: number;
}

例如：

Generate Article
maxTokens = 10000
maxCost = $0.10

超过：

budget exceeded

Job 失败或进入审批。

68. Cost Tracking

每次 Execution 记录：

inputTokens
outputTokens
totalTokens
estimatedCost
actualCost
currency

Task 汇总：

100 articles
Total tokens: 2.4M
Cost: $13.20

69. External API Usage

Execution 可以记录：

externalRequests
externalDuration
externalCost

例如：

OpenAI:
10 requests

GitHub:
5 requests

Browser:
3 sessions

用于：

Observability
Billing
Budget

70. Task Events

至少：

task.created
task.queued
task.started
task.progress
task.paused
task.resumed
task.completed
task.failed
task.cancelled

Job：

job.created
job.queued
job.started
job.retrying
job.completed
job.failed
job.cancelled

71. Task Audit

例如：

User created Task
Agent modified Task
Worker started Job
Action executed
Job retried
User cancelled Task

都进入：

Audit Runtime

72. Task API

推荐：

GET    /api/tasks
GET    /api/tasks/:id

POST   /api/tasks

POST   /api/tasks/:id/pause
POST   /api/tasks/:id/resume
POST   /api/tasks/:id/cancel
POST   /api/tasks/:id/retry

GET    /api/tasks/:id/jobs
GET    /api/tasks/:id/executions
GET    /api/tasks/:id/events

73. Job API

GET /api/jobs

GET /api/jobs/:id

POST /api/jobs/:id/retry

POST /api/jobs/:id/cancel

GET /api/jobs/:id/executions

普通用户通常不需要直接操作 Job。

Job API 主要服务：

Admin
Developer
Operations
Agent Runtime

74. Worker API

Worker 通常使用内部协议：

POST /internal/workers/register
POST /internal/workers/heartbeat
POST /internal/workers/lease
POST /internal/workers/report

也可以使用 Redis / Message Queue。

75. Queue Backend

第一版本可以使用：

Redis

实现：

Queue
Lease
Retry
Delay
Priority

数据库负责：

Task
Job
Execution
Audit

Redis 负责：

Scheduling
Queue
Locks
Heartbeat

76. Database 与 Queue 分工

不要把 Redis 当永久业务数据库。

推荐：

PostgreSQL
├── Task
├── Job
├── Execution
├── Schedule
└── Audit

Redis
├── Queue
├── Lease
├── Lock
├── Worker heartbeat
└── Temporary state

77. Queue Consistency

关键问题：

DB Job created
 ↓
Redis enqueue failed

如果直接：

DB commit
Redis enqueue

可能造成 Job 永远不执行。

因此必须支持：

Outbox Pattern

78. Outbox

流程：

DB Transaction
 ├── Create Job
 └── Create Outbox Event
          ↓
       Commit
          ↓
    Outbox Worker
          ↓
       Redis Queue

这样可以保证最终进入 Queue。

79. Event Outbox

同样：

Action Execution
 ↓
DB transaction
 ├── update execution
 └── event outbox

再：

Outbox
 ↓
Event Bus

避免：

DB 成功
Event 丢失

80. Queue At-Least-Once

默认：

At-least-once delivery

意味着：

Job 可能被执行超过一次

所以：

Action 必须尽可能幂等。

不要假设：

exactly-once

81. Duplicate Detection

Worker 获取 Job：

jobId

Action Runtime 检查：

execution
idempotency

如果已经完成：

return existing execution

82. Distributed Lock

某些 Action 必须：

single execution

例如：

production deployment

使用：

Lock:
deployment:production:app

确保：

只有一个 Job

执行。

83. Task Lock

Task 也可以：

exclusive

例如：

Database Migration

禁止同时执行：

Migration A
Migration B

84. Pause

Pause Task：

Task
 ↓
paused

行为：

queued jobs
→ 不再领取

running jobs
→ 默认继续完成

new jobs
→ 不创建/不调度

也可以提供：

hard pause

请求取消正在运行的 Job。

85. Resume

paused
 ↓
resume
 ↓
queued

继续剩余 Jobs。

已经完成的 Job：

不重复执行

86. Task Timeout

Task 可以定义：

maxDurationMs

例如：

2 hours

超过：

Task timeout

策略：

cancel remaining jobs

87. Job Timeout

每个 Job 可以定义：

timeoutMs

例如：

browser.navigate
30 sec

超时：

Execution timeout
 ↓
Abort
 ↓
Retry

如果 retryable。

88. Queue Timeout

如果 Job：

queued

超过：

maxQueueWait

可以：

fail

或者：

priority escalate

用于检测 Worker 不足。

89. Worker Auto Scaling

未来可以：

Queue Depth
 ↓
Scaler
 ↓
Worker Count

例如：

Queue:
1000 Jobs

Workers:
5

Scale:
20

但第一版本不需要复杂 Kubernetes Autoscaler。

保留接口：

interface WorkerScaler {}

90. Worker Capability

某些 Job 只能给特定 Worker。

例如：

browser Job

要求：

worker.capabilities:
browser

Server deployment：

worker.capabilities:
docker
ssh

91. Worker Tags

Worker：

tags:
browser
linux
gpu
region-us

Job：

requiredTags:
browser

Scheduler：

只分配给匹配 Worker

92. Plugin Job

Plugin 可以声明 Job 类型：

github.sync
browser.crawl
cloudflare.dns.update

但执行：

Job
 ↓
Action Runtime

Plugin 不拥有 Worker Runtime。

93. Agent Task

Agent 创建：

Task

例如：

Goal:
检查 100 个网站 SEO

Task:
seo.batch_audit

Jobs:
site-001
...
site-100

Agent：

观察
 ↓
等待 Task
 ↓
分析结果

而不是：

Agent
 ↓
100 次工具调用

94. Agent Observation

Task 完成后产生：

task.completed

Agent 收到：

Observation

然后决定：

Complete
Replan
Create another Task
Ask Human

95. Agent 与 Task 的边界

Agent:
Why / What next

Task:
Work to be done

Job:
Unit to schedule

Worker:
Who executes

Action:
How to execute

Execution:
What happened

这是整个 DevOS 最重要的边界之一。

96. Workflow 与 Task

Workflow：

预定义流程

Task：

一次实际工作

例如：

Workflow:
SEO Audit Workflow

Task:
Audit example.com

执行：

Workflow
 ↓
Task
 ↓
Jobs

97. Workflow 不应该绕过 Task

错误：

Workflow
 ↓
Action
 ↓
Action
 ↓
Action

如果流程可能：

长时间运行
可暂停
可重试
可观察

应该：

Workflow
 ↓
Task
 ↓
Jobs

98. Task UI

推荐：

/tasks

列表：

Task                    Status       Progress

SEO Audit               Running      72%

Generate Articles       Completed    100%

Deploy App              Failed       80%

GitHub Sync             Queued       0%

99. Task Detail

SEO Audit
────────────────────

Status:
Running

Progress:
72%

Jobs:
72 completed
3 failed
5 running
20 queued

Started:
10:30

Duration:
12m

Cost:
$2.31

100. Task Timeline

显示：

10:30 Task created
10:31 Job 001 started
10:31 Job 001 completed
10:32 Job 002 started
10:33 Job 002 retrying
10:34 Job 002 completed
...

数据来自：

Event
Execution
Audit

不是 UI 自己记录。

101. Job Detail

Job
────────────────

Action:
seo.website.scan

Status:
Completed

Attempts:
2

Execution:
exec_123

Worker:
worker-07

Duration:
31s

Retry:
1

102. Execution Detail

Execution
────────────────

Action:
seo.website.scan

Actor:
agent:seo-agent

Input:
example.com

Output:
...

Duration:
13.2s

External Requests:
4

Tokens:
0

Cost:
$0.00

103. Task Streaming

对于实时任务：

Task
 ↓
SSE / WebSocket
 ↓
UI

推送：

progress
logs
job status
execution output
events

104. Task Logs

统一：

Task
 ↓
Job
 ↓
Execution
 ↓
Logs

日志必须关联：

taskId
jobId
executionId
workerId
pluginId
actionId

105. Task Output

Task 可以最终产生：

interface TaskOutput {
  summary?: string;

  data?: unknown;

  artifacts?: ArtifactRef[];

  links?: LinkRef[];
}

例如：

100 SEO reports

生成：

reports/*.json
reports/*.html

通过：

Artifact Runtime

管理。

106. Artifact

不要直接把巨大结果塞进：

Task.output

例如：

1000 MB CSV

应该：

Task
 ↓
Artifact
 ↓
Object Storage

Task 只保存：

artifactId

107. Task Result Size

推荐：

small output
→ DB

large output
→ Artifact Storage

例如：

JSON < 1MB
→ DB

large HTML
→ Object Storage

具体阈值配置化。

108. Task Security

Task 必须带：

actor
organizationId
projectId

查询 Task：

User A

不能读取：

Organization B

109. Task Authorization

前端：

显示/隐藏

只是 UX。

真正：

GET /tasks/:id
POST /tasks/:id/cancel

必须经过：

Authorization
Resource Scope

110. Task Audit

Cancel：

User
 ↓
task.cancel
 ↓
Task Runtime

Audit：

actor=user
action=task.cancel
resource=task:123

111. Task API 不直接执行 Action

错误：

POST /tasks
 ↓
Controller
 ↓
executeAction()

正确：

POST /tasks
 ↓
Task Runtime
 ↓
Create Task
 ↓
Create Job
 ↓
Queue

Worker 才执行 Action。

112. Action Runtime 与 Worker

Worker 不应该自己实现 Action。

正确：

Worker
 ↓
ActionRuntime.execute(actionId, input, context)

Action Runtime 负责：

Permission
Validation
Middleware
Transaction
Audit
Execution
Error

113. Worker 与 Plugin

Worker：

不需要知道 GitHub Plugin 的内部实现

只需要：

Action ID

例如：

github.repository.sync

Action Registry：

resolve

然后执行。

114. Queue 与 Plugin

Queue 不知道：

GitHub
SEO
Browser

Queue 只知道：

Job
priority
resourceGroup
schedule

115. Scheduler 与 Agent

Scheduler 可以触发：

Agent Goal

例如：

Every Monday
 ↓
Create Agent Goal:
分析本周 SEO 数据

Agent 再：

Goal
 ↓
Plan
 ↓
Task

因此：

Scheduler
 ≠
Agent

116. Scheduler 与 Plugin

Plugin 可以声明 Schedule：

GitHub Plugin
 ↓
github.sync schedule

Scheduler 统一执行。

Plugin 不实现 Cron Worker。

117. Queue Worker Architecture

推荐：

packages/task/
├── task-runtime.ts
├── task-service.ts
├── task-progress.ts
├── task-cancellation.ts
│
├── job-runtime.ts
├── job-service.ts
├── job-factory.ts
│
├── queue/
│   ├── queue.ts
│   ├── redis-queue.ts
│   └── queue-scheduler.ts
│
├── worker/
│   ├── worker.ts
│   ├── worker-manager.ts
│   ├── worker-heartbeat.ts
│   └── worker-lease.ts
│
├── retry/
│   └── retry-policy.ts
│
├── scheduler/
│   ├── scheduler.ts
│   ├── cron.ts
│   └── triggers.ts
│
└── execution/
    └── execution-service.ts

118. Database Models

至少：

Task
Job
Execution
Schedule
Worker
JobDependency
TaskEvent
OutboxEvent

可选：

TaskArtifact
TaskLog
JobLease

119. Task Repository

interface TaskRepository {
  create(input): Promise<Task>;

  findById(id): Promise<Task | null>;

  updateStatus(id, status): Promise<void>;

  updateProgress(id, progress): Promise<void>;

  list(filter): Promise<Task[]>;
}

120. Job Repository

interface JobRepository {
  create(input): Promise<Job>;

  findById(id): Promise<Job | null>;

  claim(id, workerId): Promise<boolean>;

  updateStatus(id, status): Promise<void>;

  incrementAttempts(id): Promise<void>;

  listByTask(taskId): Promise<Job[]>;
}

121. Worker Manager

interface WorkerManager {
  register(worker): Promise<void>;

  heartbeat(workerId): Promise<void>;

  drain(workerId): Promise<void>;

  stop(workerId): Promise<void>;

  list(): Promise<Worker[]>;
}

122. Queue Adapter

不要把 Redis API 直接暴露给业务。

定义：

interface QueueAdapter {
  enqueue(job: QueueMessage): Promise<void>;

  reserve(
    queue: string,
    workerId: string
  ): Promise<QueueMessage | null>;

  ack(messageId: string): Promise<void>;

  release(messageId: string): Promise<void>;

  delay(messageId: string, delayMs: number): Promise<void>;
}

未来可以替换：

Redis
RabbitMQ
Kafka
SQS
NATS

而不修改 Task Runtime。

123. 第一版本 Queue

第一版本推荐：

PostgreSQL
+
Redis

不要一开始引入：

Kafka
RabbitMQ
Kubernetes
Temporal

除非实际规模已经需要。

架构层保留：

QueueAdapter

未来可以替换。

124. Transaction Boundary

创建 Task：

BEGIN
 ↓
Task
 ↓
Jobs
 ↓
Outbox
 ↓
COMMIT

Queue：

Outbox
 ↓
Redis

不要：

Redis enqueue
 ↓
DB transaction

否则容易产生孤儿 Job。

125. Worker Crash Recovery

场景：

Worker
 ↓
Job
 ↓
Action executing
 ↓
Worker crash

Lease：

expired

系统：

Job
 ↓
queued
 ↓
Worker B

然后：

new Execution

126. Zombie Worker

Worker 失去网络但仍运行：

Worker A

如果 Lease 已过期：

Worker B

领取 Job。

可能产生：

A + B

同时执行。

因此：

所有有副作用的 Action 必须使用 Idempotency / Distributed Lock。

127. Exactly Once 不作为基础假设

DevOS 默认：

at-least-once

开发者必须：

设计幂等

而不是依赖：

exactly-once

128. Task Retry 与 Agent Replan

两者完全不同。

Retry
=
相同 Job 再执行

Replan
=
Agent 改变解决方案

例如：

Browser API timeout

→ Retry。

而：

网站无法访问

Agent 判断：

改用 Search Plugin

→ Replan。

129. Task Failure 与 Agent Failure

Task：

failed

Agent 不一定：

failed

Agent 可以：

Observe
 ↓
Replan
 ↓
Create new Task

例如：

SEO Audit Task
 ↓
failed
 ↓
Agent:
尝试 Browser Plugin

130. Task Approval

危险 Task 可以：

waiting

等待：

Approval

例如：

Production Deployment

流程：

Agent
 ↓
Task
 ↓
Approval
 ↓
waiting
 ↓
Human approve
 ↓
queued
 ↓
Worker

131. Approval 不应该是 Worker 逻辑

Worker 遇到：

requiresApproval

不能自己：

ask user

而应该：

Task Runtime
 ↓
Approval Resource

132. Task Budget

Task 可以定义：

interface TaskBudget {
  maxJobs?: number;

  maxDurationMs?: number;

  maxCost?: number;

  maxTokens?: number;

  maxExternalRequests?: number;
}

超过预算：

budget_exceeded

133. Budget Aggregation

例如：

Task:
100 articles

Budget:
$20

Jobs：

$0.10
$0.15
$0.08
...

Task Runtime 汇总：

actualCost

超过：

pause
cancel
fail
approval

策略可配置。

134. Task Resource Limits

可以限制：

maxJobs
maxConcurrency
maxRetries
maxExecutionTime
maxQueueTime
maxCost

防止异常 Agent 产生：

1000000 Jobs

135. Task Observability

每个 Task 必须可以追踪：

Task ID
Job IDs
Execution IDs
Worker IDs
Action IDs
Plugin IDs
Events
Logs
Costs
Retries

形成完整 Trace：

Goal
 ↓
Task
 ↓
Job
 ↓
Execution
 ↓
Action
 ↓
Plugin
 ↓
External API

136. Task Trace ID

建议：

traceId

贯穿：

Agent
Task
Job
Execution
Action
Plugin
External Request

这样一次用户请求可以完整追踪。

137. Task Metrics

至少：

task.created
task.completed
task.failed

job.created
job.completed
job.failed
job.retry

queue.depth
queue.wait_time

worker.active
worker.capacity
worker.failure

execution.duration
execution.failure

138. SLO

未来可以设置：

Task queue latency
Task success rate
Job execution latency
Worker availability

例如：

95% Jobs
< 30s queue wait

139. Task Retention

Task / Job / Execution 数量可能巨大。

需要：

retention policy

例如：

Execution logs:
30 days

Task:
180 days

Audit:
根据 Audit Policy

但：

不允许未经策略自动删除 Audit。

140. Archive

旧 Task：

active DB
 ↓
archive

大数据：

PostgreSQL
 ↓
Object Storage

UI 仍然可以查询 Archive。

第一版本可以只实现：

retention metadata

暂不实现复杂归档。

141. Task Search

支持：

Task ID
Type
Status
Actor
Organization
Project
Action
Plugin
Date

例如：

plugin = github
status = failed
last 7 days

142. Admin Queue Dashboard

建议：

/admin/tasks
/admin/jobs
/admin/workers
/admin/queues

显示：

Queue Depth
Running
Failed
Retrying
Workers
Throughput
Latency

143. Queue Dashboard

Queues

default      120
ai           32
browser      8
deployment   2

点击：

Queue
 ↓
Jobs
 ↓
Workers

144. Worker Dashboard

worker-01
────────────────

Status:
Busy

Queues:
default, ai

Concurrency:
5 / 5

Jobs:
job-123
job-456
...

Heartbeat:
2 sec ago

145. Scheduler Dashboard

Schedules

GitHub Sync
Every 10 min
Next: 10:40
Healthy

SEO Report
Every Monday
Next: Monday
Enabled

146. Scheduler Misfire

如果 Scheduler 本身失败：

missed run

根据：

misfirePolicy

处理：

skip
run_once
run_all

147. Task Runtime 与 Notification

Task Event：

task.completed

Notification Plugin 可以监听：

Task completed

发送：

Email
Slack
Telegram
Web notification

因此 Task Runtime 不需要知道：

Email
Slack
Telegram

148. Task Runtime 与 Plugin

例如：

Cloudflare Plugin

提供：

cloudflare.dns.update

Task Runtime：

Task
 ↓
Job
 ↓
Action Runtime
 ↓
Cloudflare Action

149. Task Runtime 与 Resource

Task 本身也可以是：

Resource

推荐：

core.task
core.job
core.execution

这样可以复用：

Resource Registry
Action Registry
Permission
UI
Audit

150. Task Actions

至少：

core.task.create
core.task.pause
core.task.resume
core.task.cancel
core.task.retry
core.task.archive

Job：

core.job.retry
core.job.cancel

151. Task UI 自动生成

因为 Task 是 Resource：

Resource Metadata
 ↓
Generic List
 ↓
Generic Detail

可以自动生成基础 UI。

特殊任务：

Plugin

再提供：

Custom Task View

152. Task Event → Agent

例如：

task.completed

Agent Subscription：

Agent Runtime

收到：

Observation

然后：

Evaluate
 ↓
Complete / Replan

153. Task Event → Workflow

Workflow 可以：

On Task Completed

触发：

Next Task

形成：

Task A
 ↓
Event
 ↓
Workflow
 ↓
Task B

154. Task Event → Plugin

Plugin 可以监听：

task.failed

例如：

Notification Plugin

发送告警。

155. 最终完整执行模型

                         User
                           │
                         Agent
                           │
                          Goal
                           │
                        Planner
                           │
                         Task
                           │
                     ┌─────┴─────┐
                     │           │
                   Job A       Job B
                     │           │
                     └─────┬─────┘
                           │
                         Queue
                           │
                     ┌─────┴─────┐
                     │           │
                  Worker A    Worker B
                     │           │
                     └─────┬─────┘
                           │
                     Action Runtime
                           │
                    ┌──────┴──────┐
                    │             │
                 Plugin         Core
                    │             │
                    └──────┬──────┘
                           │
                       Execution
                           │
                         Event
                           │
              ┌────────────┼────────────┐
              │            │            │
             UI          Agent       Workflow

156. Zcode 实现要求

必须新增/完善：

packages/task/
packages/job/
packages/queue/
packages/worker/
packages/scheduler/
packages/execution/

如果当前项目已经存在类似模块：

不要重复创建第二套 Runtime，应将现有实现迁移/合并到统一 Runtime。

157. Zcode 禁止事项

禁止：

Plugin 自建 Queue
Plugin 自建 Worker

Agent 自建 Queue
Agent 自建 Worker

Workflow 自己执行长任务

Controller 直接执行 Action

MCP 直接执行外部业务

UI 直接调用 Plugin 内部函数

Scheduler 直接执行业务逻辑

所有执行必须最终进入：

Action Runtime

158. 第一阶段必须完成

Zcode 第一阶段：

Task CRUD
Job CRUD
Execution
Redis Queue
Worker
Retry
Progress
Cancel

必须真正可运行。

159. 第二阶段

实现：

Scheduler
Cron
Event Trigger
Job Dependency
Parallel Jobs
Batch Task
Task Budget
Concurrency
Resource Group

160. 第三阶段

实现：

Outbox
Lease Recovery
Dead Letter Queue
Replay
Advanced Retry
Worker Tags
Worker Scaling Interface
Artifact
Retention
Advanced Observability

161. 最小可运行 Demo

必须能够运行：

POST /api/tasks

创建：

Task:
Generate 10 test jobs

系统：

Task
 ↓
10 Jobs
 ↓
Redis Queue
 ↓
Worker
 ↓
Action
 ↓
10 Executions

最终：

Task:
completed

Progress:
100%

Jobs:
10 / 10

Executions:
10

162. Batch Demo

创建：

Task:
Generate 100 items

系统：

1 Task
100 Jobs
N Workers

例如：

Worker 1 → Job 1-5
Worker 2 → Job 6-10
Worker 3 → Job 11-15
...

最终：

100 completed

Agent 不参与每个 Job 的循环。

163. Retry Demo

Job：

test.retry

第一次：

fail retryable=true

自动：

retry

第二次：

success

结果：

Job:
completed

Attempts:
2

Executions:
2

164. Cancel Demo

创建：

10 long-running Jobs

用户：

Cancel Task

系统：

queued Jobs → cancelled
running Jobs → abort

最终：

Task:
cancelled

165. Scheduler Demo

创建：

Every minute

Scheduler：

Schedule
 ↓
Task
 ↓
Job
 ↓
Queue
 ↓
Worker

每分钟自动创建任务。

166. Worker Recovery Demo

启动：

Worker A

执行 Job。

强制杀掉：

Worker A

等待 Lease：

expired

启动：

Worker B

Job：

requeued

Worker B：

execute

167. 最终验收标准

以下全部成立才算完成：

[ ] Task 是统一任务模型
[ ] Job 是统一调度单位
[ ] Execution 是统一执行记录
[ ] Queue 不包含业务逻辑
[ ] Worker 不包含业务逻辑
[ ] Action Runtime 是唯一执行入口
[ ] Redis Queue 可工作
[ ] Worker Lease 可工作
[ ] Worker Heartbeat 可工作
[ ] Retry 可工作
[ ] Idempotency 可工作
[ ] Progress 可工作
[ ] Cancellation 可工作
[ ] Pause/Resume 可工作
[ ] Batch Task 可工作
[ ] Job Dependency 可工作
[ ] Scheduler 可工作
[ ] Cron 可工作
[ ] Event Trigger 可工作
[ ] Task Budget 可工作
[ ] Concurrency 可工作
[ ] Resource Group 可工作
[ ] Task Event 可工作
[ ] Audit 可工作
[ ] Agent 可以创建 Task
[ ] Agent 不直接控制 Worker
[ ] Plugin 可以创建 Task
[ ] Plugin 不创建自己的 Queue
[ ] Workflow 可以创建 Task
[ ] MCP 创建的工作进入 Task/Action Runtime
[ ] UI 可以实时观察 Task
[ ] CLI 与 UI 使用同一 Task Runtime

168. DevOS 最终执行 Golden Rule

以后 DevOS 中任何需要：

异步
批量
延迟
重试
定时
长时间
并发
后台
可恢复
可观察
可取消

的工作：

全部进入 Task Runtime。

统一模型：

Task
 ↓
Job
 ↓
Queue
 ↓
Worker
 ↓
Action Runtime
 ↓
Execution
 ↓
Event

Agent：

决定做什么

Task：

记录要做什么

Job：

定义具体工作单元

Queue：

决定什么时候做

Worker：

负责执行

Action Runtime：

负责如何安全地执行

Execution：

记录实际发生了什么

Event：

告诉系统发生了什么

这套模型必须成为 DevOS 所有 App、Plugin、MCP、Skill、Agent、Workflow、Scheduler 的唯一异步执行基础设施。

49 KiB Raw Blame History Unescape Escape

09_TASK_JOB_QUEUE_RUNTIME.md

DevOS Developer Work OS

Task / Job / Queue / Worker / Scheduler Runtime 设计规范

1. 文档目标

2. 为什么必须统一 Task Runtime

3. Task / Job / Execution 的区别

4. 核心边界

5. Task

6. Task Status

7. Task 不等于 Job

8. Job

9. Job Status

10. Execution

11. 三者关系

12. 为什么 Retry 不能创建新 Task

13. Queue

14. Queue 不执行业务逻辑

15. Queue 类型

16. Queue Priority

17. Worker

18. Worker Status

19. Worker 不应该拥有业务状态

20. Worker 执行流程

21. Lease

22. Lease Timeout

23. Heartbeat

24. Worker Graceful Shutdown

25. Worker Concurrency

26. Resource Concurrency

27. Resource Group

28. Concurrency Policy

29. Task Progress

30. Job Progress

31. Progress Event

32. Cancellation

33. AbortSignal

34. Cancellation State

35. 强制终止

36. Retry

37. Retry Example

38. Retryable Error

39. Retry 分类

40. Dead Letter Queue

41. Replay

42. Idempotency

43. Job Idempotency

44. Task Idempotency

45. Scheduling

46. Scheduler Architecture

47. Cron

48. Schedule Resource

49. Schedule Definition

50. Timezone

51. Missed Schedule

52. Event Trigger

53. Webhook Trigger

54. Task Dependencies

55. Job Dependency Graph

56. DAG

57. Parallel Jobs

58. Task Aggregation

59. Task Completion Policy

60. Batch Task

61. Batch Job Generation

62. Job Factory

63. Dynamic Job Creation

64. Dynamic Expansion 限制

65. Backpressure

66. Rate Limiting

67. Token Budget

68. Cost Tracking

69. External API Usage

70. Task Events

71. Task Audit

72. Task API

73. Job API

74. Worker API

75. Queue Backend

76. Database 与 Queue 分工

49 KiB

Raw Blame History