task-queue

# 任务队列参考 分布式任务队列系统、死信处理及熔断机制。 --- ## 任务模式 (Task Schema) { "id": "uuid", "idempotencyKey": "任务内容的哈希值", "type": "eng-backend|eng-frontend|ops-devops|...", "priority": 1-10, "dependencies": ["task-id-1", "task-id-2"], "payload": { "action": "implement|test|deploy|...", "target": "文件路径或资源", "params": {}, "goal": "成功标准（高层目标）", "constraints": ["无第三方依赖", "保持向后兼容"], "context": { "relatedFiles": ["file1.ts", "file2.ts"], "architectureDecisions": ["ADR-001: 使用 JWT 令牌"], "previousAttempts": "之前尝试过什么，为什么失败" } }, "createdAt": "ISO 时间", "claimedBy": null, "claimedAt": null, "timeout": 3600, "retries": 0, "maxRetries": 3, "backoffSeconds": 60, "lastError": null, "completedAt": null, "result": { "status": "success|failed", "output": "产出内容", "decisionReport": { ... } } } **已完成的任务必须包含决策报告 (Decision Report)。** 没有适当决策文档的任务将被标记为未完成。 --- ## 队列文件 .loki/queue/ +-- pending.json # 等待领取的任务 +-- in-progress.json # 当前正在执行的任务 +-- completed.json # 已完成的任务 +-- dead-letter.json # 失败待审查的任务 +-- cancelled.json # 已取消的任务 --- ## 队列操作 ### 领取任务 (带文件锁) python def claim_task(agent_id, agent_capabilities): with file_lock(".loki/state/locks/queue.lock", timeout=10): pending = read_json(".loki/queue/pending.json") # 查找符合条件的任务 for task in sorted(pending.tasks, key=lambda t: -t.priority): if task.type not in agent_capabilities: continue if task.claimedBy and not claim_expired(task): continue if not all_dependencies_completed(task.dependencies): continue if circuit_breaker_open(task.type): continue # 领取任务 task.claimedBy = agent_id task.claimedAt = now() move_task(task, "pending", "in-progress") return task return None ### 文件锁定 (Bash) bash #!/bin/bash # 使用 flock 进行原子任务领取 QUEUE_FILE=".loki/queue/pending.json" LOCK_FILE=".loki/state/locks/queue.lock" (