调度评分：多维可解释打分匹配 + dispatch.decision_made（Closes #9）
把派发从「能力子集 + 空闲」升级为任务为中心的多维可解释打分匹配。 新增/改动： - orchestrator/dispatch_score.py（新）：DispatchCandidate/DispatchScore + 加权掩码归一 打分（score_candidate/rank_candidates）+ build_dispatch_decision_event。 - orchestrator/main.py：抽出三模式共用的 finalize_dispatch；新增 scored_matchmake （ENABLE_DISPATCH_SCORE，默认关，与 ACO 择一）——为每个就绪任务在有能力的空闲 Agent 间按 capability/历史成功(τ)/负载/预算压力/权限择优，记录可解释决策；_run_budget_pressure 计算真实预算占比。 - orchestrator/swarm_runtime.py：SwarmRun.dispatch_decisions + record_dispatch_decision （内部状态，非 Manager 事件）。 - 测试：scripts/test-dispatch-score.py（公式 24 项）+ scripts/test-dispatch-scored.py （集成 11 项：按 τ/负载多 Agent 择优 + 排除原因 + 可回放记录）；CI 纳入两者。 - docs/scheduling/dispatch-score-schema.md、CLAUDE.md 同步。 诚实边界：risk_score/estimated_cost/estimated_time 本仓无来源 → None 并在 payload uncollected_dimensions 披露（不伪造，规则 #9）；dispatch.decision_made 暂为 Swarm 内部 记录，未进 Manager 事件契约（需 event-schema 注册，跨端）。flag 关闭时贪心/ACO 路径逐字节不变。 Closes #9 Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Songhaoz666
co-authored by
Claude Opus 4.8
Claude Opus 4.8
parent
423cc6fdc7
commit
62610a7e3f
@@ -86,3 +86,11 @@ jobs:
|
||||
- name: End-to-end workflow test (ACO dispatch on)
|
||||
env: { REDIS_FAKE: "1", ENABLE_ACO_DISPATCH: "1", ACO_SEED: "42" }
|
||||
run: python scripts/test-workflow-e2e.py
|
||||
|
||||
- name: Dispatch scoring formulas (#9 unit)
|
||||
env: { REDIS_FAKE: "1" }
|
||||
run: python scripts/test-dispatch-score.py
|
||||
|
||||
- name: Dispatch scoring integration (#9, scored matchmaking)
|
||||
env: { REDIS_FAKE: "1", ENABLE_DISPATCH_SCORE: "1" }
|
||||
run: python scripts/test-dispatch-scored.py
|
||||
|
||||
@@ -22,8 +22,9 @@
|
||||
## 架构与关键约束（便于定位）
|
||||
- **orchestrator/**：FastAPI 编排器。Manager 面接口、HMAC 签名回调、审批链**必须保持契约**。Redis 为权威存储；内存回退仅限 `REDIS_FAKE` / `ALLOW_MEMORY_STORE`（开发/CI）。
|
||||
- **agent/**：执行单元，**OpenAI 兼容**模型；保留计费/审计归属（`usage` 与 `X-Agent/X-Agnet` 头）。
|
||||
- **工作流开关默认关闭**：`ENABLE_PLANNER_FALLBACK`、`ENABLE_REVIEW_LOOP`、`ENABLE_SUBTASK_HANDOFF`、`ENABLE_QUALITY_EVAL`、`ENABLE_ACO_DISPATCH`。
|
||||
- **工作流开关默认关闭**：`ENABLE_PLANNER_FALLBACK`、`ENABLE_REVIEW_LOOP`、`ENABLE_SUBTASK_HANDOFF`、`ENABLE_QUALITY_EVAL`、`ENABLE_ACO_DISPATCH`、`ENABLE_DISPATCH_SCORE`。
|
||||
- **ACO 决策引擎**（`orchestrator/decision_engine.py`）：信息素**学习常开**（被动观察，不改行为）；**概率选择仅在** `ENABLE_ACO_DISPATCH=1` 时生效（改变派发顺序，CI 用 `ACO_SEED` 固定随机数）。设计见 `docs/benchmark/decision-engine.md`。
|
||||
- **调度评分**（`orchestrator/dispatch_score.py`，issue #9）：`ENABLE_DISPATCH_SCORE=1` 启用**任务为中心的可解释打分匹配**——为每个就绪任务按能力/历史成功(τ)/负载/预算压力/权限等≥4 个非能力维度给候选 Agent 打分并择优，产出 `dispatch.decision_made` 内部记录（候选明细+排除原因，存 `SwarmRun.dispatch_decisions`，**非 Manager 事件**，可审计/回放）。默认关；与 `ENABLE_ACO_DISPATCH` 互斥（择一）。设计见 `docs/scheduling/dispatch-score-schema.md`。
|
||||
- **代码测试沙箱**（`orchestrator/sandbox.py`）：会**执行模型生成代码**，OS 级隔离边界 = K8s Pod。**Fail-closed 双门控**：需同时 `ENABLE_QUALITY_EVAL=1`（开功能）+ `HEICODE_SANDBOX_ISOLATED=1`（显式确认运行在隔离 Pod）；二者缺一则启动拒绝/运行时抛 `SandboxIsolationError`，不执行任何代码。安全模型见 `docs/integration/security-boundary.md §8.1`。**`HEICODE_SANDBOX_ISOLATED` 只允许在真正隔离的 Pod 或 ephemeral CI/test runner 中设置。**
|
||||
- 提交前必须本地通过：
|
||||
```
|
||||
|
||||
@@ -0,0 +1,156 @@
|
||||
# Dispatch 评分模型与 `dispatch.decision_made` 事件（Issue #9）
|
||||
|
||||
让派发从「仅按能力集合匹配」升级为「多维、可解释、加权」的调度评分：在能力之外引入**负载、预算压力、风险、权限契合、历史质量、预估成本/时间**等维度，并把每次派发的候选列表、被选 Agent、逐维度评分细分与排除原因，通过 `dispatch.decision_made` 事件暴露出来。本文档是该评分层的唯一入口。
|
||||
|
||||
实现：`orchestrator/dispatch_score.py`（纯评分）+ `orchestrator/main.py: scored_matchmake/finalize_dispatch`（已接入派发环，门控 `ENABLE_DISPATCH_SCORE`，默认关）。测试：`scripts/test-dispatch-score.py`（公式）+ `scripts/test-dispatch-scored.py`（集成）。详见 §4。
|
||||
|
||||
## 1. 与 #10（τ/η/P 决策）的边界
|
||||
|
||||
#9 与 #10（`orchestrator/decision_engine.py`，已实现）耦合但分工明确：
|
||||
|
||||
| Issue | 产出 | 性质 |
|
||||
|---|---|---|
|
||||
| **#10** | 把历史归一为 τ、把先验归一为 η，按 `P=τ^α·η^β/Σ` **概率采样**一个任务（ε-greedy），并产出可回放的内部 `DecisionTrace` | 概率决策、含 RNG、内部状态（不进 Manager 事件流） |
|
||||
| **#9（本文）** | 候选 Agent×任务的**可解释评分输入**（capability_match / load / budget_pressure / risk_score / permission_fit / historical_success / estimated_cost / estimated_time）+ 每次派发的候选列表与排除原因（`dispatch.decision_made` 事件） | 确定性（无 RNG）、可审计、面向 Manager |
|
||||
|
||||
- **共享输入**：`historical_success` 直接复用 #10 的 τ trail（τ 是跨 run 学到的「实测声誉」，已在 `[0.05, 1.0]`），经 `normalize_tau` 归一到 `[0,1]`。`capability_match`、`load` 与 `DecisionEngine.compute_eta` 的 `match`/`resource` 同定义，保证两层对「能力契合」「负载」理解一致。
|
||||
- **`dispatch_score.py` 模块本身**是纯计算、无 RNG、不发事件。**选择行为**由 `main.py` 的集成提供：开 `ENABLE_DISPATCH_SCORE` 时，`scored_matchmake` 用本评分**确定性地在多个候选 Agent 中择优**（任务为中心，与 #10 的概率采样互斥、择一启用）。即 #9 = 确定性可解释打分匹配；#10 = 概率采样。两者都默认关，关闭时为原贪心派发。
|
||||
|
||||
## 2. 评分模型
|
||||
|
||||
### 2.1 维度与权重（`DISPATCH_WEIGHTS`）
|
||||
|
||||
| 维度 | 权重 | 方向 | 含义 / 归一来源 |
|
||||
|---|---|---|---|
|
||||
| `capability_match` | 0.30 | 收益 | required ∩ caps 的 **Jaccard**（奖励专精）；无 required → 1.0（任何人可做，对齐 `can_agent_run_task`） |
|
||||
| `historical_success` | 0.20 | 收益 | 复用 #10 的 τ trail，`normalize_tau` 映射到 `[0,1]` |
|
||||
| `load` | 0.15 | 收益 | 空闲槽位余量（越空越高），对齐 `AGENT_SLOTS` / `RESOURCE_NORM_SLOTS` |
|
||||
| `permission_fit` | 0.12 | 收益 | Agent 是否被允许执行该任务（1.0 允许 / 0.0 不允许） |
|
||||
| `budget_pressure` | 0.10 | **成本** | run 预算已消耗比例（越高越扣分） |
|
||||
| `risk_score` | 0.08 | **成本** | 任务风险（越高越扣分） |
|
||||
| `estimated_cost` | 0.03 | **成本** | 预估 $（越高越扣分） |
|
||||
| `estimated_time` | 0.02 | **成本** | 预估墙钟（越高越扣分） |
|
||||
|
||||
- 权重显式、可审计，**不必和为 1.0**——评分时按「实际有信号的维度」重归一（见 §2.2）。
|
||||
- 成本维度（`budget_pressure`/`risk_score`/`estimated_cost`/`estimated_time`）的贡献取 `(1 − 值)`，即「越高越差」；其余为收益维度，贡献取原值。
|
||||
|
||||
### 2.2 打分公式（掩码 + 重归一）
|
||||
|
||||
```
|
||||
total = Σ_{d∈present} w_d · contribution_d / Σ_{d∈present} w_d
|
||||
contribution_d = (1 − x_d) if d 是成本维度
|
||||
= x_d 否则 （x_d 夹紧到 [0,1]）
|
||||
```
|
||||
|
||||
- **值为 `None` 的维度同时退出分子与权重归一**（masked + renormalized），因此「未采集」的信号既不加分也不扣分，`total` 始终落在 `[0,1]`。这与 `benchmark.metrics.quality_score` 的掩码纪律一致（规则 #9：绝不为未采集量伪造 0/1）。
|
||||
- 权重为 0 的维度仍写入 `breakdown`（透明），但不参与计分。
|
||||
|
||||
### 2.3 数据结构
|
||||
|
||||
- `DispatchCandidate`：一个 (agent, task) 配对的原始信号。除标识字段外每个字段都是评分输入；**无信号的维度必须留 `None`，不得伪造**。`score` 由 `score_candidate()` 回填。
|
||||
- `DispatchScore`：
|
||||
- `total`：重归一加权总分（`[0,1]`）；
|
||||
- `breakdown`：每个维度的**原始值**（或 `None`），供审计；
|
||||
- `weights_used`：实际施加到各「在场」维度的重归一权重；
|
||||
- `missing`：本次无信号的维度名列表（显式列出「未采集」集合，而非静默省略）。
|
||||
|
||||
### 2.4 函数
|
||||
|
||||
| 函数 | 职责 |
|
||||
|---|---|
|
||||
| `normalize_capability_match(required, caps)` | Jaccard；无 required → 1.0 |
|
||||
| `normalize_tau(tau)` | τ → `[0,1]`；`None → None` |
|
||||
| `normalize_load(free_slots)` | 空闲槽位 → `[0,1]`；`None → None` |
|
||||
| `score_candidate(candidate)` | 纯函数打分（无 RNG），回填 `candidate.score` |
|
||||
| `rank_candidates(candidates)` | 全部打分并按 total 降序（稳定排序，决定性） |
|
||||
| `build_dispatch_decision_event(candidates, chosen, excluded_reasons, task_id=...)` | 构造 `dispatch.decision_made` 事件 payload |
|
||||
|
||||
## 3. `dispatch.decision_made` 事件
|
||||
|
||||
### 3.1 payload 结构
|
||||
|
||||
```jsonc
|
||||
{
|
||||
"task_id": "swarm-X-impl", // 与 envelope 顶层 task_id 一致
|
||||
"chosen_agent_id": "agent-A", // 被选 Agent（无派发时为 null）
|
||||
"chosen_task_id": "swarm-X-impl",
|
||||
"candidate_count": 3,
|
||||
"candidates": [ // 全部候选 + 逐维度细分
|
||||
{
|
||||
"agent_id": "agent-A",
|
||||
"task_id": "swarm-X-impl",
|
||||
"agent_role": "implementation",
|
||||
"score": {
|
||||
"total": 0.83,
|
||||
"breakdown": { // 原始值（或 null）
|
||||
"capability_match": 1.0,
|
||||
"historical_success": 0.95,
|
||||
"load": 1.0,
|
||||
"permission_fit": 1.0,
|
||||
"budget_pressure": 0.10,
|
||||
"risk_score": null,
|
||||
"estimated_cost": null,
|
||||
"estimated_time": null
|
||||
},
|
||||
"weights_used": { "capability_match": 0.30, "historical_success": 0.20, "load": 0.15, "permission_fit": 0.12, "budget_pressure": 0.10 },
|
||||
"missing": ["risk_score", "estimated_cost", "estimated_time"]
|
||||
}
|
||||
}
|
||||
// ... 其余候选
|
||||
],
|
||||
"excluded": { // agent_id → 排除原因（人读）
|
||||
"agent-B": "lower_score",
|
||||
"agent-C": "capability_mismatch"
|
||||
},
|
||||
"weights": { /* DISPATCH_WEIGHTS 快照 */ },
|
||||
"uncollected_dimensions": ["estimated_cost", "estimated_time", "risk_score"]
|
||||
}
|
||||
```
|
||||
|
||||
### 3.2 Envelope 与事件流约定
|
||||
|
||||
- 本 payload 由 orchestrator 的 `swarm_runtime.emit_event(run, "dispatch.decision_made", task_id=..., payload=...)` 包裹，自动补齐 `event_id/swarm_id/occurred_at/correlation_id/source` 等标准 envelope 字段（见 `docs/integration/event-schema.md §2`）。
|
||||
- **该事件类型当前不在 `event-schema.md §4` 的 HM 注册表中**，属于 Swarm 内部可观测扩展。是否需要 HM 入库/前端展示需与 Manager 侧对齐——在对齐前，事件只进 Swarm 本地事件流，不应宣称已被 HM 校验/消费（见 §5）。
|
||||
|
||||
## 4. 集成现状（已落地）
|
||||
|
||||
> **已接入 `orchestrator/main.py`**（issue #9 关闭 PR）。下述为实际实现，非提案。
|
||||
|
||||
启用开关 **`ENABLE_DISPATCH_SCORE`（默认关，与 `ENABLE_ACO_DISPATCH` 择一）**。开启后，`task_dispatch_loop` 在每个 tick 走 **`scored_matchmake(...)` 任务为中心的匹配**，而非 Agent 拉取：
|
||||
|
||||
1. 枚举所有就绪 PENDING 任务（按 `created_at` 稳定排序）。
|
||||
2. 对每个任务，遍历**有能力且有容量**的空闲 Agent，构造 `DispatchCandidate`，信号取自既有状态：
|
||||
- `capability_match` = `normalize_capability_match(required, caps)`（Jaccard）
|
||||
- `historical_success` = `normalize_tau(decision_engine.get_tau(role, agent))`（**复用 #10 的 τ trail**）
|
||||
- `load` = `normalize_load(AGENT_SLOTS[agent])`（空闲槽位）
|
||||
- `permission_fit` = 1.0（`can_agent_run_task` 通过；不通过的 Agent 直接以 `capability_mismatch` 进 `excluded`，不参与排序）
|
||||
- `budget_pressure` = **真实计算** `_run_budget_pressure(task)`（run 已消耗成本 / `max_cost_usd`；无预算或无用量 → None）
|
||||
- `risk_score` / `estimated_cost` / `estimated_time` = None（本仓无来源，见 §5，payload `uncollected_dimensions` 披露）
|
||||
3. `rank_candidates(...)` 确定性择优；落选 Agent 记 `lower_score`，无能力 Agent 记 `capability_mismatch`。
|
||||
4. 选中对经 **`finalize_dispatch(agent, task, dispatch_event=...)`**（三种派发模式共用的统一收尾）完成 assign + `task.claimed` + 派发上下文 + 下发；同一 tick 内被选 Agent 不重复分配。
|
||||
5. `dispatch.decision_made` payload（候选明细 + 排除原因）经 **`swarm_runtime.record_dispatch_decision`** 存入 **`SwarmRun.dispatch_decisions`（内部状态，可审计/回放）**。
|
||||
|
||||
> **刻意不发 Manager 事件**：`dispatch.decision_made` 暂为 **Swarm 内部可观测记录**，**未**经 `swarm_runtime.emit_event` 进入回调流——避免污染未注册的 Manager 事件契约。若 HM 需消费，需先在 `docs/integration/event-schema.md §4` 注册（跨端，超出 #9 范围）。`build_dispatch_decision_event` 与 `dispatch_score_event_enabled()` 已备好，供注册后启用 Manager 侧发送。
|
||||
>
|
||||
> 行为隔离：`ENABLE_DISPATCH_SCORE` 关闭时，贪心/ACO 路径**逐字节不变**（共用的 `finalize_dispatch` 保持原语义）。
|
||||
>
|
||||
> 测试：`scripts/test-dispatch-score.py`（纯公式 24 项）+ `scripts/test-dispatch-scored.py`（集成：按 τ/负载在多 Agent 中择优 + 记录可回放，11 项）。运行命令见各测试 docstring 与 `docs/TESTING.md`。
|
||||
|
||||
## 5. 诚实边界（本仓**未**采集 / **未**做的）
|
||||
|
||||
严格遵循规则 #9，以下维度在本仓**无真实来源**，一律以 `None`（"not collected"）表示，**绝不伪造数值**；并通过 payload 的 `uncollected_dimensions` 显式披露：
|
||||
|
||||
- **`estimated_cost` / `estimated_time`**：本仓无「派发前」成本/时长预估器。模型成本与 `runtime_seconds` 只在任务**完成后**经 `emit_usage_event` 得到（事后量），派发时不可得 → `None`。
|
||||
- **`risk_score`**：本仓无每任务风险分类器。风险/审批（`risk_level`）归属 **Manager 审批链**，不在队列 `Task` 上 → `None`。
|
||||
- **region / GPU / 硬件契合**：`AgentMetadata` 无任何此类字段（仅 `agent_id/status/last_heartbeat/capabilities/current_task_id`）→ **不引入该维度**（不造假维度）。
|
||||
- **`budget_pressure`**：理论上可由 run 预算与已消耗比例算出，但派发前缺少可靠的「run 级已消耗」累计来源（usage 事件按任务事后发出），故集成示例中留 `None`；若后续聚合 run 级已消耗，可填真实比例。
|
||||
- **`permission_fit`**：当前**仅能**用「能力子集」近似（`can_agent_run_task`）。真正的 RBAC/审批级权限校验在 Manager 侧，Swarm 派发面无此信号——本维度是「能力可行性」的近似，不是完整权限判定。
|
||||
- **决策质量未证**：本评分维度/权重是否能产出更优派发是**经验命题**，需 Group C 的对比 harness 才能回答；在此之前，本评分层默认不发事件（`ENABLE_DISPATCH_SCORE_EVENT` 关）、不改派发行为。
|
||||
|
||||
## 6. 文件
|
||||
|
||||
| 文件 | 职责 |
|
||||
|---|---|
|
||||
| `orchestrator/dispatch_score.py` | `DispatchCandidate`/`DispatchScore` 数据结构、归一函数、`score_candidate`/`rank_candidates`、`build_dispatch_decision_event` |
|
||||
| `scripts/test-dispatch-score.py` | 24 项断言（归一器、同能力不同代价/负载/历史 → 不同分且可解释、≥4 个非能力维度各自影响评分、None 掩码不扣分、事件携带候选细分 + 排除原因 + 未采集披露） |
|
||||
| `docs/scheduling/dispatch-score-schema.md` | 本文档 |
|
||||
@@ -0,0 +1,344 @@
|
||||
"""Explainable dispatch scoring layer (issue #9).
|
||||
|
||||
This module produces the EXPLAINABLE SCORE INPUTS for capability-aware swarm
|
||||
dispatch and the `dispatch.decision_made` event that surfaces them. It answers
|
||||
"why was this agent×task pairing chosen, and why were the others excluded?" with
|
||||
a per-dimension breakdown — going beyond the previous behavior, which matched on
|
||||
capability set membership alone (`TaskQueue.can_agent_run_task`).
|
||||
|
||||
Boundary with #10 (`orchestrator/decision_engine.py`, already implemented):
|
||||
- #10 = the PROBABILISTIC decision: normalize history into τ and a-priori
|
||||
desirability into η, then SAMPLE one task with P=τ^α·η^β/Σ (ε-greedy), and
|
||||
emit a replayable internal `DecisionTrace`.
|
||||
- #9 (this module) = the EXPLAINABLE SCORE INPUTS + the Manager-facing
|
||||
`dispatch.decision_made` event. It computes a transparent, weighted,
|
||||
per-dimension score for each candidate and records exclusion reasons. It is
|
||||
deterministic (no RNG) and does not itself change dispatch order.
|
||||
- Shared input: `historical_success` reuses the decision_engine τ trail where
|
||||
natural (τ is the cross-run learned reputation, already on [0.05, 1.0]).
|
||||
|
||||
Honesty (org rule #9 — no fabricated signals):
|
||||
Several scheduling dimensions named in the standard have NO source in this
|
||||
repo. They are represented as `None` ("not collected"), never invented:
|
||||
- estimated_cost / estimated_time: no pre-execution estimator exists; model
|
||||
cost/runtime are only known AFTER a task completes (see emit_usage_event in
|
||||
main.py). Pre-dispatch they are None.
|
||||
- risk_score: no per-task risk classifier. Approval/risk_level lives on the
|
||||
Manager approval chain, not on the queue Task. None here.
|
||||
- region / GPU / hardware fit: AgentMetadata carries no such fields. Not
|
||||
collected — not represented at all (we do not add a fake dimension).
|
||||
A `None` dimension contributes 0 to the weighted score AND drops out of the
|
||||
weight normalizer (masked + renormalized), so it neither helps nor penalizes a
|
||||
candidate and the score stays on [0, 1]. This mirrors metrics.quality_score().
|
||||
|
||||
Default-OFF: this module is pure computation and emits no event by itself. The
|
||||
dispatch loop only calls it when ENABLE_DISPATCH_SCORE_EVENT is set (see the
|
||||
flag helper below and the integration notes in
|
||||
docs/scheduling/dispatch-score-schema.md). With the flag off, behavior is
|
||||
byte-for-byte unchanged.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
from dataclasses import asdict, dataclass, field
|
||||
from typing import Dict, List, Optional, Sequence
|
||||
|
||||
# Weight of each scoring dimension. Positive dimensions reward a candidate;
|
||||
# *_pressure / risk are costs the caller passes already oriented as "higher =
|
||||
# worse" and we subtract them. Weights are explicit and documented so the score
|
||||
# is auditable; they need not sum to 1.0 because we renormalize over the
|
||||
# dimensions that actually have a signal (see score_candidate).
|
||||
DISPATCH_WEIGHTS: Dict[str, float] = {
|
||||
"capability_match": 0.30, # how well agent caps cover task requirements
|
||||
"historical_success": 0.20, # reuse decision_engine τ (learned reputation)
|
||||
"load": 0.15, # free-slot headroom (more free → higher)
|
||||
"permission_fit": 0.12, # agent is allowed to run this task
|
||||
"budget_pressure": 0.10, # run budget consumed so far (cost → subtract)
|
||||
"risk_score": 0.08, # task risk (cost → subtract)
|
||||
"estimated_cost": 0.03, # predicted $ (cost → subtract)
|
||||
"estimated_time": 0.02, # predicted wall-clock (cost → subtract)
|
||||
}
|
||||
|
||||
# Dimensions treated as costs: a higher value LOWERS the score. The rest are
|
||||
# benefits: a higher value RAISES it.
|
||||
COST_DIMENSIONS = frozenset({"budget_pressure", "risk_score", "estimated_cost", "estimated_time"})
|
||||
|
||||
DISPATCH_DECISION_EVENT_TYPE = "dispatch.decision_made"
|
||||
|
||||
|
||||
def dispatch_score_event_enabled() -> bool:
|
||||
"""Whether the dispatch loop should emit `dispatch.decision_made` (default OFF).
|
||||
|
||||
Pure scoring/explanation is always safe to compute; this flag only gates the
|
||||
new event emission so the Manager event stream is unchanged by default.
|
||||
"""
|
||||
return os.getenv("ENABLE_DISPATCH_SCORE_EVENT", "false").lower() in {"1", "true", "yes"}
|
||||
|
||||
|
||||
@dataclass
|
||||
class DispatchScore:
|
||||
"""Explainable, weighted score for one (agent, task) pairing.
|
||||
|
||||
`total` is the renormalized weighted blend of the per-dimension values in
|
||||
`breakdown`. `breakdown` keeps the RAW dimension values (each on [0, 1], or
|
||||
None when not collected) so a reviewer can see exactly what drove the total.
|
||||
`weights_used` records the renormalized weight actually applied to each
|
||||
present dimension (absent dimensions are dropped), and `missing` lists the
|
||||
dimensions that had no signal — making the "not collected" set explicit in
|
||||
the payload rather than silently omitted.
|
||||
"""
|
||||
total: float
|
||||
breakdown: Dict[str, Optional[float]]
|
||||
weights_used: Dict[str, float]
|
||||
missing: List[str] = field(default_factory=list)
|
||||
|
||||
def as_dict(self) -> Dict[str, object]:
|
||||
return {
|
||||
"total": round(self.total, 6),
|
||||
"breakdown": {k: (round(v, 6) if isinstance(v, (int, float)) else None)
|
||||
for k, v in self.breakdown.items()},
|
||||
"weights_used": {k: round(v, 6) for k, v in self.weights_used.items()},
|
||||
"missing": list(self.missing),
|
||||
}
|
||||
|
||||
|
||||
@dataclass
|
||||
class DispatchCandidate:
|
||||
"""One (agent, task) pairing considered for dispatch, with its raw signals.
|
||||
|
||||
Every field except the identifiers is an explainable input to the score.
|
||||
Fields whose signal does not exist in this repo MUST be left None by the
|
||||
caller — they are not invented (org rule #9). See module docstring for the
|
||||
list of inherently-None dimensions (estimated_cost/time, risk_score, region/
|
||||
GPU). `score` is filled in by score_candidate().
|
||||
"""
|
||||
agent_id: str
|
||||
task_id: str
|
||||
agent_role: str = "general"
|
||||
# --- benefit dimensions (higher = better), each normalized to [0, 1] ---
|
||||
capability_match: float = 0.0 # Jaccard / coverage of required caps
|
||||
historical_success: Optional[float] = None # decision_engine τ, normalized
|
||||
load: Optional[float] = None # free-slot headroom, normalized
|
||||
permission_fit: Optional[float] = None # 1.0 allowed, 0.0 disallowed
|
||||
# --- cost dimensions (higher = worse), each normalized to [0, 1] ---
|
||||
budget_pressure: Optional[float] = None
|
||||
risk_score: Optional[float] = None
|
||||
estimated_cost: Optional[float] = None
|
||||
estimated_time: Optional[float] = None
|
||||
# filled by score_candidate
|
||||
score: Optional[DispatchScore] = None
|
||||
|
||||
def signal_dimensions(self) -> Dict[str, Optional[float]]:
|
||||
"""Return the raw scoring inputs keyed by dimension name."""
|
||||
return {
|
||||
"capability_match": self.capability_match,
|
||||
"historical_success": self.historical_success,
|
||||
"load": self.load,
|
||||
"permission_fit": self.permission_fit,
|
||||
"budget_pressure": self.budget_pressure,
|
||||
"risk_score": self.risk_score,
|
||||
"estimated_cost": self.estimated_cost,
|
||||
"estimated_time": self.estimated_time,
|
||||
}
|
||||
|
||||
|
||||
def _clamp01(x: float) -> float:
|
||||
return 0.0 if x < 0.0 else (1.0 if x > 1.0 else x)
|
||||
|
||||
|
||||
def normalize_capability_match(required: Sequence[str], capabilities: Sequence[str]) -> float:
|
||||
"""Jaccard overlap of required vs. agent caps — rewards focus, matches η.
|
||||
|
||||
Aligned with DecisionEngine.compute_eta's `match` (Jaccard rewards focus),
|
||||
with one deliberate refinement matching TaskQueue.can_agent_run_task:
|
||||
a task with NO required capabilities can be run by anyone → perfect fit
|
||||
(1.0), rather than the raw Jaccard 0 you would get for required=∅.
|
||||
"""
|
||||
required_set = set(required or [])
|
||||
if not required_set:
|
||||
return 1.0
|
||||
caps_set = set(capabilities or [])
|
||||
union = required_set | caps_set
|
||||
if not union:
|
||||
return 1.0
|
||||
return len(required_set & caps_set) / len(union)
|
||||
|
||||
|
||||
def normalize_tau(tau: Optional[float], *, tau_min: float = 0.05, tau_max: float = 1.0) -> Optional[float]:
|
||||
"""Map a decision_engine τ trail value onto [0, 1] for historical_success.
|
||||
|
||||
τ already lives on [TAU_MIN, TAU_MAX]; we rescale so the breakdown is on the
|
||||
same [0,1] axis as the other dimensions. None in → None out (no trail yet is
|
||||
"not collected", not "zero reputation" — do not fabricate).
|
||||
"""
|
||||
if tau is None:
|
||||
return None
|
||||
span = tau_max - tau_min
|
||||
if span <= 0:
|
||||
return _clamp01(tau)
|
||||
return _clamp01((tau - tau_min) / span)
|
||||
|
||||
|
||||
def normalize_load(free_slots: Optional[int], *, saturation_slots: float = 2.0) -> Optional[float]:
|
||||
"""Map free capacity onto [0, 1] (more headroom → higher). None → None.
|
||||
|
||||
Mirrors DecisionEngine RESOURCE_NORM_SLOTS so load means the same thing in
|
||||
both layers.
|
||||
"""
|
||||
if free_slots is None:
|
||||
return None
|
||||
if saturation_slots <= 0:
|
||||
return 1.0 if free_slots > 0 else 0.0
|
||||
return _clamp01(max(0, free_slots) / saturation_slots)
|
||||
|
||||
|
||||
def score_candidate(
|
||||
candidate: DispatchCandidate,
|
||||
*,
|
||||
weights: Optional[Dict[str, float]] = None,
|
||||
) -> DispatchScore:
|
||||
"""Compute the explainable, weighted score for one candidate (pure, no RNG).
|
||||
|
||||
Score = Σ_{d∈present} w_d · contribution_d / Σ_{d∈present} w_d, where a
|
||||
benefit dimension contributes its value and a cost dimension contributes
|
||||
(1 − value). Dimensions whose value is None are masked out of BOTH the
|
||||
numerator and the weight normalizer, so an uncollected signal neither helps
|
||||
nor hurts and the result stays on [0, 1]. This is the same masking discipline
|
||||
as benchmark.metrics.quality_score (org rule #9 — no fabricated 0/1).
|
||||
|
||||
All present dimensions are also written back into the returned breakdown so
|
||||
the choice is fully auditable, and `missing` names the absent ones.
|
||||
"""
|
||||
w = weights or DISPATCH_WEIGHTS
|
||||
raw = candidate.signal_dimensions()
|
||||
|
||||
numerator = 0.0
|
||||
weight_total = 0.0
|
||||
weights_used: Dict[str, float] = {}
|
||||
missing: List[str] = []
|
||||
|
||||
for dim, value in raw.items():
|
||||
if value is None:
|
||||
missing.append(dim)
|
||||
continue
|
||||
weight = w.get(dim, 0.0)
|
||||
if weight == 0.0:
|
||||
# Recorded in breakdown for transparency but carries no score weight.
|
||||
continue
|
||||
clamped = _clamp01(float(value))
|
||||
contribution = (1.0 - clamped) if dim in COST_DIMENSIONS else clamped
|
||||
numerator += weight * contribution
|
||||
weight_total += weight
|
||||
weights_used[dim] = weight
|
||||
|
||||
total = (numerator / weight_total) if weight_total > 0 else 0.0
|
||||
score = DispatchScore(
|
||||
total=total,
|
||||
breakdown=dict(raw),
|
||||
weights_used=weights_used,
|
||||
missing=missing,
|
||||
)
|
||||
candidate.score = score
|
||||
return score
|
||||
|
||||
|
||||
def rank_candidates(
|
||||
candidates: Sequence[DispatchCandidate],
|
||||
*,
|
||||
weights: Optional[Dict[str, float]] = None,
|
||||
) -> List[DispatchCandidate]:
|
||||
"""Score every candidate and return them sorted best-first (stable).
|
||||
|
||||
Pure helper for callers/tests that want the explainable ranking without the
|
||||
probabilistic sampling of decision_engine. Ties keep input order (stable
|
||||
sort), so the result is deterministic.
|
||||
"""
|
||||
for candidate in candidates:
|
||||
if candidate.score is None:
|
||||
score_candidate(candidate, weights=weights)
|
||||
return sorted(candidates, key=lambda c: c.score.total, reverse=True)
|
||||
|
||||
|
||||
def build_dispatch_decision_event(
|
||||
candidates: Sequence[DispatchCandidate],
|
||||
chosen: Optional[DispatchCandidate],
|
||||
excluded_reasons: Optional[Dict[str, str]] = None,
|
||||
*,
|
||||
task_id: Optional[str] = None,
|
||||
weights: Optional[Dict[str, float]] = None,
|
||||
) -> Dict[str, object]:
|
||||
"""Build a `dispatch.decision_made` event payload (issue #9 DoD).
|
||||
|
||||
The payload carries:
|
||||
- `candidates`: every considered (agent, task) pairing with its full
|
||||
per-dimension score breakdown (raw values + renormalized weights +
|
||||
the explicit `missing` list of uncollected dimensions);
|
||||
- `chosen_agent_id` / `chosen_task_id`: the selected pairing (or None when
|
||||
nothing dispatched);
|
||||
- `excluded`: agent_id → human-readable reason a candidate was NOT chosen
|
||||
(e.g. "capability_mismatch", "no_capacity", "lower_score").
|
||||
|
||||
This is an EXPLANATION of dispatch inputs, not the probabilistic decision
|
||||
itself (#10 owns that and stores its own replayable DecisionTrace
|
||||
internally). The payload follows the event-schema.md envelope convention:
|
||||
the orchestrator's emit_event wraps this under `payload` and adds the
|
||||
standard envelope fields (event_id/swarm_id/occurred_at/...). The caller
|
||||
supplies `task_id` for the envelope's top-level task_id field too — see the
|
||||
integration notes in docs/scheduling/dispatch-score-schema.md.
|
||||
|
||||
Any candidate not yet scored is scored here, so the payload is always
|
||||
self-consistent.
|
||||
"""
|
||||
excluded = dict(excluded_reasons or {})
|
||||
scored: List[Dict[str, object]] = []
|
||||
for candidate in candidates:
|
||||
score = candidate.score or score_candidate(candidate, weights=weights)
|
||||
scored.append({
|
||||
"agent_id": candidate.agent_id,
|
||||
"task_id": candidate.task_id,
|
||||
"agent_role": candidate.agent_role,
|
||||
"score": score.as_dict(),
|
||||
})
|
||||
|
||||
resolved_task_id = task_id or (chosen.task_id if chosen else None)
|
||||
if resolved_task_id is None and candidates:
|
||||
resolved_task_id = candidates[0].task_id
|
||||
|
||||
return {
|
||||
"task_id": resolved_task_id,
|
||||
"chosen_agent_id": chosen.agent_id if chosen else None,
|
||||
"chosen_task_id": chosen.task_id if chosen else None,
|
||||
"candidate_count": len(scored),
|
||||
"candidates": scored,
|
||||
"excluded": excluded,
|
||||
"weights": dict(weights or DISPATCH_WEIGHTS),
|
||||
# The dimensions that are inherently None in this repo (see module
|
||||
# docstring). Surfaced so the Manager UI can show "not collected" rather
|
||||
# than assume the scorer ignored them.
|
||||
"uncollected_dimensions": _repo_uncollected_dimensions(),
|
||||
}
|
||||
|
||||
|
||||
def _repo_uncollected_dimensions() -> List[str]:
|
||||
"""Dimensions with NO data source in this repo today (honest disclosure).
|
||||
|
||||
These are reported on every event so downstream consumers never mistake a
|
||||
masked dimension for a low score. Kept in code (not a doc-only note) so the
|
||||
disclosure ships with the payload.
|
||||
"""
|
||||
return ["estimated_cost", "estimated_time", "risk_score"]
|
||||
|
||||
|
||||
__all__ = [
|
||||
"DispatchCandidate",
|
||||
"DispatchScore",
|
||||
"DISPATCH_WEIGHTS",
|
||||
"DISPATCH_DECISION_EVENT_TYPE",
|
||||
"dispatch_score_event_enabled",
|
||||
"normalize_capability_match",
|
||||
"normalize_tau",
|
||||
"normalize_load",
|
||||
"score_candidate",
|
||||
"rank_candidates",
|
||||
"build_dispatch_decision_event",
|
||||
]
|
||||
+149
-37
@@ -27,6 +27,10 @@ from .planner import planner
|
||||
from .master_agent import master_agent
|
||||
from .quality import evaluate_run_quality
|
||||
from .decision_engine import decision_engine, aco_dispatch_enabled
|
||||
from .dispatch_score import (
|
||||
DispatchCandidate, rank_candidates, build_dispatch_decision_event,
|
||||
normalize_capability_match, normalize_tau, normalize_load, dispatch_score_event_enabled,
|
||||
)
|
||||
|
||||
# Configure logging
|
||||
logging.basicConfig(
|
||||
@@ -304,6 +308,16 @@ async def task_dispatch_loop():
|
||||
and agent_has_capacity(agent.agent_id)
|
||||
]
|
||||
|
||||
# Issue #9: task-centric scored matchmaking — for each ready task, score the capable
|
||||
# idle agents on ≥4 non-capability dimensions and assign the best, recording an
|
||||
# explainable dispatch.decision_made record. Gated (default OFF); when off, the
|
||||
# agent-centric greedy/ACO paths below are byte-for-byte unchanged.
|
||||
if dispatch_score_enabled():
|
||||
for agent, task, event_payload in await scored_matchmake(connected_idle_agents):
|
||||
await task_queue.remove_pending_task(task.task_id)
|
||||
await finalize_dispatch(agent, task, dispatch_event=event_payload)
|
||||
continue
|
||||
|
||||
for agent in connected_idle_agents:
|
||||
decision = None
|
||||
if aco_dispatch_enabled():
|
||||
@@ -331,48 +345,146 @@ async def task_dispatch_loop():
|
||||
if not task:
|
||||
break
|
||||
|
||||
success = await task_queue.assign_task(task.task_id, agent.agent_id)
|
||||
if not success:
|
||||
latest_task = await task_queue.get_task(task.task_id)
|
||||
if latest_task and latest_task.status == TaskStatus.PENDING:
|
||||
await task_queue.requeue_task(task.task_id)
|
||||
continue
|
||||
|
||||
run = await swarm_runtime.get_run_for_task(task.task_id)
|
||||
if run:
|
||||
await swarm_runtime.emit_event(
|
||||
run,
|
||||
"task.claimed",
|
||||
task_id=task.task_id,
|
||||
agent_instance_id=agent.agent_id,
|
||||
payload={
|
||||
"task_id": task.task_id,
|
||||
"agent_role": task.agent_role,
|
||||
"agent_id": agent.agent_id,
|
||||
},
|
||||
)
|
||||
if decision is not None:
|
||||
# Group A telemetry: feeds tau/eta/p_decision in the collector.
|
||||
await swarm_runtime.record_decision(run, decision.telemetry())
|
||||
|
||||
dispatch_context = (
|
||||
await build_dispatch_context(run, task) if run else task.context
|
||||
)
|
||||
await manager.send_message(agent.agent_id, {
|
||||
"type": "task_assignment",
|
||||
"task_id": task.task_id,
|
||||
"description": task.description,
|
||||
"context": dispatch_context,
|
||||
})
|
||||
|
||||
logger.info(
|
||||
f"Dispatched task {task.task_id} to connected idle agent {agent.agent_id}"
|
||||
)
|
||||
await finalize_dispatch(agent, task, decision=decision)
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Error in task dispatch loop: {e}")
|
||||
|
||||
|
||||
async def finalize_dispatch(agent, task, *, decision=None, dispatch_event=None) -> bool:
|
||||
"""Shared dispatch tail used by all selection modes (greedy / ACO / scored).
|
||||
|
||||
Assigns the task, emits task.claimed, records the selection telemetry (Group A decision
|
||||
and/or #9 dispatch decision — both INTERNAL run state, not Manager events), builds the
|
||||
dispatch context, and sends the assignment. Returns False (and requeues) if the assign lost
|
||||
a race. Extracted so the three modes never diverge.
|
||||
"""
|
||||
success = await task_queue.assign_task(task.task_id, agent.agent_id)
|
||||
if not success:
|
||||
latest_task = await task_queue.get_task(task.task_id)
|
||||
if latest_task and latest_task.status == TaskStatus.PENDING:
|
||||
await task_queue.requeue_task(task.task_id)
|
||||
return False
|
||||
|
||||
run = await swarm_runtime.get_run_for_task(task.task_id)
|
||||
if run:
|
||||
await swarm_runtime.emit_event(
|
||||
run,
|
||||
"task.claimed",
|
||||
task_id=task.task_id,
|
||||
agent_instance_id=agent.agent_id,
|
||||
payload={
|
||||
"task_id": task.task_id,
|
||||
"agent_role": task.agent_role,
|
||||
"agent_id": agent.agent_id,
|
||||
},
|
||||
)
|
||||
if decision is not None:
|
||||
# Group A telemetry: feeds tau/eta/p_decision in the collector.
|
||||
await swarm_runtime.record_decision(run, decision.telemetry())
|
||||
if dispatch_event is not None:
|
||||
# #9: internal dispatch-decision record (audit/benchmark-replayable). NOT a Manager
|
||||
# event — a Manager-facing dispatch.decision_made needs event-schema registration.
|
||||
await swarm_runtime.record_dispatch_decision(run, dispatch_event)
|
||||
|
||||
dispatch_context = await build_dispatch_context(run, task) if run else task.context
|
||||
await manager.send_message(agent.agent_id, {
|
||||
"type": "task_assignment",
|
||||
"task_id": task.task_id,
|
||||
"description": task.description,
|
||||
"context": dispatch_context,
|
||||
})
|
||||
logger.info(f"Dispatched task {task.task_id} to connected idle agent {agent.agent_id}")
|
||||
return True
|
||||
|
||||
|
||||
def dispatch_score_enabled() -> bool:
|
||||
return os.getenv("ENABLE_DISPATCH_SCORE", "false").lower() in {"1", "true", "yes"}
|
||||
|
||||
|
||||
async def _run_budget_pressure(task) -> Optional[float]:
|
||||
"""Fraction of the run budget already consumed (real signal for dispatch scoring), or None.
|
||||
|
||||
Run-level (same across agents for a task; it prioritizes tasks on cheaper runs). None when
|
||||
the run has no budget or no usage yet — never fabricated.
|
||||
"""
|
||||
run = await swarm_runtime.get_run_for_task(task.task_id)
|
||||
if not run:
|
||||
return None
|
||||
budget = ((run.request_body or {}).get("orchestration_plan") or {}).get("budget") or {}
|
||||
max_cost = budget.get("max_cost_usd")
|
||||
if not isinstance(max_cost, (int, float)) or max_cost <= 0:
|
||||
return None
|
||||
consumed = 0.0
|
||||
for tid in run.task_ids:
|
||||
t = await task_queue.get_task(tid)
|
||||
if not t or not t.result:
|
||||
continue
|
||||
try:
|
||||
data = json.loads(t.result) if isinstance(t.result, str) else t.result
|
||||
consumed += float((data.get("usage") or {}).get("model_cost_usd") or 0.0)
|
||||
except Exception:
|
||||
continue
|
||||
return max(0.0, min(1.0, consumed / float(max_cost)))
|
||||
|
||||
|
||||
async def scored_matchmake(idle_agents):
|
||||
"""Issue #9: task-centric explainable scored dispatch.
|
||||
|
||||
For each ready pending task, score every capable, has-capacity idle agent on capability_match
|
||||
+ historical_success (decision_engine τ) + load (free slots) + permission_fit + budget_pressure
|
||||
(≥4 non-capability dimensions with real in-repo signals; risk/estimated_cost/estimated_time
|
||||
have no source → None, disclosed), pick the highest-scoring agent, and build the
|
||||
dispatch.decision_made payload (candidate breakdown + exclusion reasons). Deterministic. An
|
||||
agent chosen for one task is not reused in the same tick. Returns [(agent, task, event)].
|
||||
"""
|
||||
assignments = []
|
||||
used_agents: set = set()
|
||||
all_tasks = await task_queue.get_all_tasks()
|
||||
ready = [t for t in all_tasks
|
||||
if t.status == TaskStatus.PENDING and await task_queue.is_task_ready(t)]
|
||||
ready.sort(key=lambda t: t.created_at) # stable, oldest-first
|
||||
|
||||
for task in ready:
|
||||
required = task.required_capabilities or []
|
||||
budget_pressure = await _run_budget_pressure(task)
|
||||
candidates = [] # DispatchCandidate for permitted agents
|
||||
cand_agent = {} # agent_id -> agent
|
||||
excluded = {} # agent_id -> reason
|
||||
for agent in idle_agents:
|
||||
if agent.agent_id in used_agents or not agent_has_capacity(agent.agent_id):
|
||||
continue
|
||||
caps = set(agent.capabilities or [])
|
||||
permitted = task_queue.can_agent_run_task(task, caps)
|
||||
if not permitted:
|
||||
excluded[agent.agent_id] = "capability_mismatch"
|
||||
continue
|
||||
tau = await decision_engine.get_tau(task.agent_role, agent.agent_id)
|
||||
cand = DispatchCandidate(
|
||||
agent_id=agent.agent_id,
|
||||
task_id=task.task_id,
|
||||
agent_role=task.agent_role,
|
||||
capability_match=normalize_capability_match(required, caps),
|
||||
historical_success=normalize_tau(tau),
|
||||
load=normalize_load(AGENT_SLOTS.get(agent.agent_id, 1)),
|
||||
permission_fit=1.0,
|
||||
budget_pressure=budget_pressure,
|
||||
)
|
||||
candidates.append(cand)
|
||||
cand_agent[agent.agent_id] = agent
|
||||
if not candidates:
|
||||
continue
|
||||
ranked = rank_candidates(candidates)
|
||||
chosen = ranked[0]
|
||||
for cand in ranked[1:]:
|
||||
excluded[cand.agent_id] = "lower_score"
|
||||
event = build_dispatch_decision_event(candidates, chosen, excluded, task_id=task.task_id)
|
||||
chosen_agent = cand_agent[chosen.agent_id]
|
||||
used_agents.add(chosen_agent.agent_id)
|
||||
assignments.append((chosen_agent, task, event))
|
||||
return assignments
|
||||
|
||||
|
||||
async def refresh_swarm_run_status(run):
|
||||
"""Update a swarm run once all known tasks have reached terminal states."""
|
||||
if run.status == "stopped":
|
||||
|
||||
@@ -58,6 +58,12 @@ class SwarmRun(BaseModel):
|
||||
# p_norm, p_score, explored}). Internal state (NOT a Manager event). Appended only when
|
||||
# ENABLE_ACO_DISPATCH selected the assignment; empty → tau/eta/p_decision stay NaN.
|
||||
decisions: List[Dict[str, Any]] = Field(default_factory=list)
|
||||
# Issue #9: explainable dispatch-scoring decisions (`dispatch.decision_made` payloads:
|
||||
# candidate agents + per-dimension score breakdown + exclusion reasons). Internal state
|
||||
# (NOT a Manager event — a Manager-facing dispatch event needs event-schema registration,
|
||||
# cross-team). Appended only when ENABLE_DISPATCH_SCORE selected the assignment. Audit/
|
||||
# benchmark-replayable from here.
|
||||
dispatch_decisions: List[Dict[str, Any]] = Field(default_factory=list)
|
||||
metadata: Dict[str, Any] = Field(default_factory=dict)
|
||||
request_body: Dict[str, Any] = Field(default_factory=dict)
|
||||
created_at: float = Field(default_factory=time.time)
|
||||
@@ -378,6 +384,11 @@ class SwarmRuntime:
|
||||
run.decisions = (run.decisions or [])[-999:] + [decision]
|
||||
await self.save_run(run)
|
||||
|
||||
async def record_dispatch_decision(self, run: SwarmRun, decision: Dict[str, Any]) -> None:
|
||||
"""Append one explainable dispatch-scoring decision (#9). Internal, bounded list."""
|
||||
run.dispatch_decisions = (run.dispatch_decisions or [])[-999:] + [decision]
|
||||
await self.save_run(run)
|
||||
|
||||
async def attach_task(self, run: SwarmRun, task_id: str):
|
||||
"""Associate a queue task with a swarm run."""
|
||||
if task_id not in run.task_ids:
|
||||
|
||||
@@ -0,0 +1,163 @@
|
||||
"""Tests for the explainable dispatch scoring layer (issue #9).
|
||||
|
||||
Hermetic and deterministic: this exercises orchestrator/dispatch_score.py, which
|
||||
is pure computation (no RNG, no Redis, no WebSocket, no model key). It does not
|
||||
touch the dispatch loop or the decision engine's sampling — it only verifies the
|
||||
explainable score INPUTS and the `dispatch.decision_made` event payload.
|
||||
|
||||
Run from agent_swarm_v6 (install deps first to match the other test scripts;
|
||||
this test itself imports nothing beyond the stdlib + dispatch_score, so it also
|
||||
runs with no deps installed):
|
||||
pip install -r orchestrator/requirements.txt -r agent/requirements.txt
|
||||
..\\.venv\\Scripts\\python.exe scripts/test-dispatch-score.py
|
||||
"""
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
|
||||
|
||||
from orchestrator.dispatch_score import (
|
||||
DISPATCH_DECISION_EVENT_TYPE,
|
||||
DispatchCandidate,
|
||||
build_dispatch_decision_event,
|
||||
normalize_capability_match,
|
||||
normalize_load,
|
||||
normalize_tau,
|
||||
rank_candidates,
|
||||
score_candidate,
|
||||
)
|
||||
|
||||
failures = []
|
||||
|
||||
|
||||
def check(name, cond):
|
||||
print(("PASS" if cond else "FAIL"), "-", name)
|
||||
if not cond:
|
||||
failures.append(name)
|
||||
|
||||
|
||||
def main():
|
||||
# --- normalizers map real signals onto [0,1] and preserve "not collected" ---
|
||||
check("capability Jaccard rewards focus",
|
||||
normalize_capability_match(["python"], ["python"])
|
||||
> normalize_capability_match(["python"], ["python", "go", "rust", "java"]))
|
||||
check("no requirements -> perfect capability fit",
|
||||
normalize_capability_match([], ["python"]) == 1.0)
|
||||
check("tau normalized into [0,1]", 0.0 <= normalize_tau(0.5) <= 1.0)
|
||||
check("tau None stays None (not collected, not zero)", normalize_tau(None) is None)
|
||||
check("more free slots -> higher load score", normalize_load(2) > normalize_load(0))
|
||||
check("load None stays None", normalize_load(None) is None)
|
||||
|
||||
# --- SAME capability, DIFFERENT cost/load/historical_success -> different scores ---
|
||||
# All three share capability_match=1.0 and permission_fit=1.0; only the
|
||||
# non-capability dimensions vary. A correct scorer must separate them.
|
||||
base = dict(task_id="t1", agent_role="implementation",
|
||||
capability_match=1.0, permission_fit=1.0)
|
||||
strong = DispatchCandidate(agent_id="A", historical_success=0.95, load=1.0,
|
||||
budget_pressure=0.1, **base)
|
||||
weak = DispatchCandidate(agent_id="B", historical_success=0.10, load=0.2,
|
||||
budget_pressure=0.9, **base)
|
||||
s_strong = score_candidate(strong)
|
||||
s_weak = score_candidate(weak)
|
||||
check("same capability+permission -> different total by other dims",
|
||||
abs(s_strong.total - s_weak.total) > 1e-6)
|
||||
check("better history/load/budget scores higher", s_strong.total > s_weak.total)
|
||||
check("score stays on [0,1]", 0.0 <= s_strong.total <= 1.0 and 0.0 <= s_weak.total <= 1.0)
|
||||
|
||||
# --- the chosen one is EXPLAINABLE: ranking + breakdown justify it ---
|
||||
ranked = rank_candidates([weak, strong])
|
||||
chosen = ranked[0]
|
||||
check("ranking picks the stronger candidate", chosen.agent_id == "A")
|
||||
check("breakdown carries every dimension",
|
||||
{"capability_match", "historical_success", "load", "permission_fit",
|
||||
"budget_pressure", "risk_score", "estimated_cost", "estimated_time"}
|
||||
<= set(chosen.score.breakdown))
|
||||
# cost dimension is inverted internally: lower budget_pressure must help, and
|
||||
# we can see it in the raw breakdown.
|
||||
check("chosen has lower budget_pressure recorded",
|
||||
chosen.score.breakdown["budget_pressure"] < s_weak.breakdown["budget_pressure"])
|
||||
|
||||
# --- non-capability dimensions REALLY affect the score (>=4 of them) ---
|
||||
# Flip one non-capability dimension at a time from a neutral baseline and
|
||||
# confirm each independently moves the total. capability_match is held fixed.
|
||||
def neutral():
|
||||
return DispatchCandidate(
|
||||
agent_id="N", task_id="t1", capability_match=1.0, permission_fit=0.5,
|
||||
historical_success=0.5, load=0.5, budget_pressure=0.5,
|
||||
risk_score=0.5, estimated_cost=0.5, estimated_time=0.5,
|
||||
)
|
||||
|
||||
baseline_total = score_candidate(neutral()).total
|
||||
movers = []
|
||||
for dim, better_value in [
|
||||
("historical_success", 1.0), # benefit up -> score up
|
||||
("load", 1.0), # benefit up -> score up
|
||||
("permission_fit", 1.0), # benefit up -> score up
|
||||
("budget_pressure", 0.0), # cost down -> score up
|
||||
("risk_score", 0.0), # cost down -> score up
|
||||
("estimated_cost", 0.0), # cost down -> score up
|
||||
("estimated_time", 0.0), # cost down -> score up
|
||||
]:
|
||||
cand = neutral()
|
||||
setattr(cand, dim, better_value)
|
||||
moved = score_candidate(cand).total
|
||||
if moved > baseline_total + 1e-9:
|
||||
movers.append(dim)
|
||||
check("at least 4 non-capability dimensions independently raise the score",
|
||||
len(movers) >= 4)
|
||||
check("each of >=4 movers is a distinct non-capability dimension",
|
||||
len(set(movers)) >= 4 and "capability_match" not in movers)
|
||||
|
||||
# --- honesty: None dimensions are masked, not fabricated to 0 ---
|
||||
# A candidate with several signals uncollected must NOT be punished for them:
|
||||
# masking out a None dimension should give the SAME total as a candidate that
|
||||
# only has the collected dimensions.
|
||||
partial = DispatchCandidate(agent_id="P", task_id="t1", capability_match=1.0,
|
||||
permission_fit=1.0, historical_success=0.8)
|
||||
sp = score_candidate(partial)
|
||||
check("uncollected dims listed in `missing`",
|
||||
{"load", "budget_pressure", "risk_score",
|
||||
"estimated_cost", "estimated_time"} <= set(sp.missing))
|
||||
check("masked None dims do not drag score toward 0", sp.total > 0.5)
|
||||
check("weights_used only counts present dims",
|
||||
set(sp.weights_used) == {"capability_match", "historical_success", "permission_fit"})
|
||||
|
||||
# --- event payload carries candidate breakdown + exclusion reasons ---
|
||||
excluded = {"B": "lower_score", "C": "capability_mismatch"}
|
||||
mismatch = DispatchCandidate(agent_id="C", task_id="t1", capability_match=0.0,
|
||||
permission_fit=0.0)
|
||||
score_candidate(mismatch)
|
||||
event = build_dispatch_decision_event(
|
||||
candidates=[strong, weak, mismatch],
|
||||
chosen=strong,
|
||||
excluded_reasons=excluded,
|
||||
task_id="t1",
|
||||
)
|
||||
check("event type constant matches schema", DISPATCH_DECISION_EVENT_TYPE == "dispatch.decision_made")
|
||||
check("event records chosen agent + task",
|
||||
event["chosen_agent_id"] == "A" and event["chosen_task_id"] == "t1")
|
||||
check("event lists all candidates", event["candidate_count"] == 3 and len(event["candidates"]) == 3)
|
||||
check("each event candidate carries a full score breakdown",
|
||||
all("breakdown" in c["score"] and "missing" in c["score"] and "weights_used" in c["score"]
|
||||
for c in event["candidates"]))
|
||||
check("event carries exclusion reasons", event["excluded"] == excluded)
|
||||
check("event discloses uncollected dimensions honestly",
|
||||
set(event["uncollected_dimensions"]) == {"estimated_cost", "estimated_time", "risk_score"})
|
||||
|
||||
# --- no dispatch case: chosen=None is representable ---
|
||||
empty_event = build_dispatch_decision_event(candidates=[mismatch], chosen=None,
|
||||
excluded_reasons={"C": "capability_mismatch"})
|
||||
check("no-dispatch event has chosen None but still explains the excluded candidate",
|
||||
empty_event["chosen_agent_id"] is None
|
||||
and empty_event["excluded"] == {"C": "capability_mismatch"}
|
||||
and empty_event["candidate_count"] == 1)
|
||||
|
||||
print()
|
||||
if failures:
|
||||
print(f"{len(failures)} dispatch-score check(s) FAILED: {failures}")
|
||||
sys.exit(1)
|
||||
print("all dispatch-score checks passed")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -0,0 +1,117 @@
|
||||
"""Integration test for issue #9: task-centric SCORED dispatch picks among AGENTS.
|
||||
|
||||
Exercises orchestrator.main.scored_matchmake against the real task_queue + agent_registry +
|
||||
decision_engine (REDIS_FAKE, no WS/model): two capable agents differ in historical success (τ)
|
||||
and load; the higher-scored agent is chosen; an incapable agent is excluded; the explainable
|
||||
dispatch.decision_made record (candidate breakdown + exclusion reasons + ≥4 non-capability
|
||||
dimensions with real signals) is produced and stored on the run.
|
||||
|
||||
Run from agent_swarm_v6 (install deps first — needs fakeredis):
|
||||
pip install -r orchestrator/requirements.txt -r agent/requirements.txt
|
||||
REDIS_FAKE=1 python scripts/test-dispatch-scored.py
|
||||
"""
|
||||
import asyncio
|
||||
import os
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
os.environ["REDIS_FAKE"] = "1"
|
||||
os.environ["ENABLE_DISPATCH_SCORE"] = "1"
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
|
||||
|
||||
from orchestrator.redis_client import redis_client
|
||||
from orchestrator import swarm_runtime as sr_mod
|
||||
from orchestrator.swarm_runtime import swarm_runtime
|
||||
from orchestrator.task_queue import task_queue, TaskStatus
|
||||
from orchestrator.agent_registry import agent_registry
|
||||
from orchestrator.decision_engine import decision_engine
|
||||
from orchestrator import main as orch
|
||||
|
||||
failures = []
|
||||
|
||||
|
||||
def check(name, cond):
|
||||
print(("PASS" if cond else "FAIL"), "-", name)
|
||||
if not cond:
|
||||
failures.append(name)
|
||||
|
||||
|
||||
async def _noop(self, *a, **k):
|
||||
return None
|
||||
|
||||
|
||||
async def main():
|
||||
await redis_client.connect()
|
||||
sr_mod.SwarmRuntime._post_callback = _noop
|
||||
|
||||
body = {
|
||||
"mode": "swarm",
|
||||
"requirement": {"objective": "scored dispatch test"},
|
||||
"orchestration_plan": {"budget": {"max_cost_usd": 10}},
|
||||
"callback": {"url": "http://127.0.0.1:9/cb", "subscribed_events": []},
|
||||
"metadata": {"manager_deployment_id": "m-disp"},
|
||||
}
|
||||
run, _ = await swarm_runtime.get_or_create_run(body=body, idempotency_key=None, correlation_id="cd")
|
||||
|
||||
# One ready task requiring python.
|
||||
task = await task_queue.create_task(task_id="t-implementation", description="impl",
|
||||
agent_role="implementation",
|
||||
required_capabilities=["python"], enqueue=True)
|
||||
await swarm_runtime.attach_task(run, task.task_id)
|
||||
|
||||
# Three agents: A & B both capable (python); C incapable (java only).
|
||||
await agent_registry.register_agent("agent-A", ["python"])
|
||||
await agent_registry.register_agent("agent-B", ["python"])
|
||||
await agent_registry.register_agent("agent-C", ["java"])
|
||||
orch.AGENT_SLOTS.update({"agent-A": 1, "agent-B": 1, "agent-C": 1})
|
||||
|
||||
# Historical success (τ) differs: A strong, B weak — same capability & load otherwise.
|
||||
for _ in range(8):
|
||||
await decision_engine.deposit(agent_role="implementation", agent_id="agent-A", success=True)
|
||||
await decision_engine.deposit(agent_role="implementation", agent_id="agent-B", success=False)
|
||||
|
||||
idle = [a for a in await agent_registry.get_idle_agents()
|
||||
if a.agent_id in {"agent-A", "agent-B", "agent-C"}]
|
||||
assignments = await orch.scored_matchmake(idle)
|
||||
|
||||
check("exactly one assignment produced", len(assignments) == 1)
|
||||
agent, chosen_task, event = assignments[0]
|
||||
check("higher-τ agent A chosen over B (historical_success decides)", agent.agent_id == "agent-A")
|
||||
check("chosen task is the python task", chosen_task.task_id == "t-implementation")
|
||||
|
||||
# Explainable event payload
|
||||
check("event names chosen agent/task",
|
||||
event["chosen_agent_id"] == "agent-A" and event["chosen_task_id"] == "t-implementation")
|
||||
check("incapable agent C excluded as capability_mismatch",
|
||||
event["excluded"].get("agent-C") == "capability_mismatch")
|
||||
check("loser B excluded as lower_score", event["excluded"].get("agent-B") == "lower_score")
|
||||
a_cand = next(c for c in event["candidates"] if c["agent_id"] == "agent-A")
|
||||
present = {k for k, v in a_cand["score"]["breakdown"].items() if v is not None}
|
||||
noncap_present = present - {"capability_match"}
|
||||
check("≥4 non-capability dimensions have real signals",
|
||||
{"historical_success", "load", "permission_fit", "budget_pressure"} <= noncap_present
|
||||
and len(noncap_present) >= 4)
|
||||
check("uncollected dims disclosed (no fabrication)",
|
||||
set(event["uncollected_dimensions"]) >= {"estimated_cost", "estimated_time", "risk_score"})
|
||||
check("A's historical_success > B's in breakdown",
|
||||
a_cand["score"]["breakdown"]["historical_success"]
|
||||
> next(c for c in event["candidates"] if c["agent_id"] == "agent-B")["score"]["breakdown"]["historical_success"])
|
||||
|
||||
# finalize_dispatch stores the decision on the run (audit/benchmark-replayable, internal)
|
||||
await task_queue.remove_pending_task(chosen_task.task_id)
|
||||
ok = await orch.finalize_dispatch(agent, chosen_task, dispatch_event=event)
|
||||
check("finalize_dispatch assigned the task", ok is True)
|
||||
refreshed = await swarm_runtime.get_run(run.swarm_id)
|
||||
check("dispatch decision recorded on run (replayable)",
|
||||
len(refreshed.dispatch_decisions) == 1
|
||||
and refreshed.dispatch_decisions[0]["chosen_agent_id"] == "agent-A")
|
||||
|
||||
print()
|
||||
if failures:
|
||||
print(f"{len(failures)} scored-dispatch check(s) FAILED: {failures}")
|
||||
sys.exit(1)
|
||||
print("all scored-dispatch (#9) checks passed")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
asyncio.run(main())
|
||||
Reference in New Issue
Block a user