Compare commits

...

1 Commits

Author SHA1 Message Date
Alexander Whitestone
0b4b20f62e feat: add fleet dispatch planning scaffold (#552)
Some checks are pending
Smoke Test / smoke (pull_request) Waiting to run
2026-04-15 00:48:41 -04:00
2 changed files with 237 additions and 20 deletions

View File

@@ -1,31 +1,169 @@
#!/usr/bin/env python3
import json
import os
import yaml
from pathlib import Path
"""Dynamic dispatch optimizer for fleet-wide coordination.
# Dynamic Dispatch Optimizer
# Automatically updates routing based on fleet health.
Refs: timmy-home #552
Takes a fleet dispatch spec plus optional failover status and produces a
capacity-aware assignment plan. Safe by default: it prints the plan and only
writes an output file when explicitly requested.
"""
from __future__ import annotations
import argparse
import json
from pathlib import Path
from typing import Any
STATUS_FILE = Path.home() / ".timmy" / "failover_status.json"
CONFIG_FILE = Path.home() / "timmy" / "config.yaml"
SPEC_FILE = Path.home() / ".timmy" / "fleet_dispatch.json"
OUTPUT_FILE = Path.home() / ".timmy" / "dispatch_plan.json"
def load_json(path: Path, default: Any):
if not path.exists():
return default
return json.loads(path.read_text())
def _host_status(host: dict[str, Any], failover_status: dict[str, Any]) -> str:
if host.get("always_available"):
return "ONLINE"
fleet = failover_status.get("fleet") or {}
return str(fleet.get(host["name"], "ONLINE")).upper()
def _lane_matches(host: dict[str, Any], lane: str) -> bool:
host_lanes = set(host.get("lanes") or ["general"])
if host.get("always_available", False):
return True
if lane == "general":
return "general" in host_lanes
return lane in host_lanes
def _choose_candidate(task: dict[str, Any], hosts: list[dict[str, Any]]):
lane = task.get("lane", "general")
preferred = task.get("preferred_hosts") or []
preferred_map = {host["name"]: host for host in hosts}
for host_name in preferred:
host = preferred_map.get(host_name)
if not host:
continue
if host["remaining_capacity"] <= 0:
continue
if _lane_matches(host, lane):
return host
matching = [host for host in hosts if host["remaining_capacity"] > 0 and _lane_matches(host, lane)]
if matching:
matching.sort(key=lambda host: (host["assigned_count"], -host["remaining_capacity"], host["name"]))
return matching[0]
fallbacks = [host for host in hosts if host["remaining_capacity"] > 0 and host.get("always_available")]
if fallbacks:
fallbacks.sort(key=lambda host: (host["assigned_count"], -host["remaining_capacity"], host["name"]))
return fallbacks[0]
return None
def generate_plan(spec: dict[str, Any], failover_status: dict[str, Any] | None = None) -> dict[str, Any]:
failover_status = failover_status or {}
raw_hosts = spec.get("hosts") or []
tasks = list(spec.get("tasks") or [])
online_hosts = []
offline_hosts = []
for host in raw_hosts:
normalized = {
"name": host["name"],
"capacity": int(host.get("capacity", 1)),
"remaining_capacity": int(host.get("capacity", 1)),
"assigned_count": 0,
"lanes": list(host.get("lanes") or ["general"]),
"always_available": bool(host.get("always_available", False)),
"status": _host_status(host, failover_status),
}
if normalized["status"] == "ONLINE":
online_hosts.append(normalized)
else:
offline_hosts.append(normalized["name"])
ordered_tasks = sorted(
tasks,
key=lambda item: (-int(item.get("priority", 0)), str(item.get("id", ""))),
)
assignments = []
unassigned = []
for task in ordered_tasks:
candidate = _choose_candidate(task, online_hosts)
if candidate is None:
unassigned.append({
"task_id": task.get("id"),
"reason": f"no_online_host_for_lane:{task.get('lane', 'general')}",
})
continue
candidate["remaining_capacity"] -= 1
candidate["assigned_count"] += 1
assignments.append({
"task_id": task.get("id"),
"host": candidate["name"],
"lane": task.get("lane", "general"),
"priority": int(task.get("priority", 0)),
})
return {
"assignments": assignments,
"offline_hosts": sorted(offline_hosts),
"unassigned": unassigned,
}
def write_plan(plan: dict[str, Any], output_path: Path):
output_path.parent.mkdir(parents=True, exist_ok=True)
output_path.write_text(json.dumps(plan, indent=2))
def parse_args():
parser = argparse.ArgumentParser(description="Generate a fleet dispatch plan from host health and task demand.")
parser.add_argument("--spec-file", type=Path, default=SPEC_FILE, help="JSON fleet spec with hosts[] and tasks[]")
parser.add_argument("--status-file", type=Path, default=STATUS_FILE, help="Failover monitor JSON payload")
parser.add_argument("--output", type=Path, default=OUTPUT_FILE, help="Output path for the generated plan")
parser.add_argument("--write-output", action="store_true", help="Persist the generated plan to --output")
parser.add_argument("--json", action="store_true", help="Print JSON only")
return parser.parse_args()
def main():
print("--- Allegro's Dynamic Dispatch Optimizer ---")
if not STATUS_FILE.exists():
print("No failover status found.")
args = parse_args()
spec = load_json(args.spec_file, {"hosts": [], "tasks": []})
failover_status = load_json(args.status_file, {})
plan = generate_plan(spec, failover_status)
if args.write_output:
write_plan(plan, args.output)
if args.json:
print(json.dumps(plan, indent=2))
return
status = json.loads(STATUS_FILE.read_text())
fleet = status.get("fleet", {})
# Logic: If primary VPS is offline, switch fallback to local Ollama
if fleet.get("ezra") == "OFFLINE":
print("Ezra (Primary) is OFFLINE. Optimizing for local-only fallback...")
# In a real scenario, this would update the YAML config
print("Updated config.yaml: fallback_model -> ollama:gemma4:12b")
else:
print("Fleet health is optimal. Maintaining high-performance routing.")
print("--- Dynamic Dispatch Optimizer ---")
print(f"Assignments: {len(plan['assignments'])}")
if plan["offline_hosts"]:
print("Offline hosts: " + ", ".join(plan["offline_hosts"]))
for assignment in plan["assignments"]:
print(f"- {assignment['task_id']} -> {assignment['host']} ({assignment['lane']}, p={assignment['priority']})")
if plan["unassigned"]:
print("Unassigned:")
for item in plan["unassigned"]:
print(f"- {item['task_id']}: {item['reason']}")
if args.write_output:
print(f"Wrote plan to {args.output}")
if __name__ == "__main__":
main()

View File

@@ -0,0 +1,79 @@
import json
from scripts.dynamic_dispatch_optimizer import generate_plan, write_plan
def test_generate_plan_rebalances_offline_host_tasks_to_online_capacity():
spec = {
"hosts": [
{"name": "ezra", "capacity": 2, "lanes": ["research", "general"]},
{"name": "bezalel", "capacity": 2, "lanes": ["build", "general"]},
{"name": "local", "capacity": 1, "lanes": ["general"], "always_available": True},
],
"tasks": [
{"id": "ISSUE-1", "lane": "build", "priority": 100},
{"id": "ISSUE-2", "lane": "general", "priority": 80},
{"id": "ISSUE-3", "lane": "research", "priority": 60},
],
}
failover_status = {"fleet": {"ezra": "ONLINE", "bezalel": "OFFLINE"}}
plan = generate_plan(spec, failover_status)
assignments = {item["task_id"]: item["host"] for item in plan["assignments"]}
assert assignments == {
"ISSUE-1": "local",
"ISSUE-2": "ezra",
"ISSUE-3": "ezra",
}
assert plan["offline_hosts"] == ["bezalel"]
assert plan["unassigned"] == []
def test_generate_plan_prefers_preferred_host_when_online():
spec = {
"hosts": [
{"name": "ezra", "capacity": 2, "lanes": ["general"]},
{"name": "bezalel", "capacity": 2, "lanes": ["general"]},
],
"tasks": [
{"id": "ISSUE-9", "lane": "general", "priority": 100, "preferred_hosts": ["bezalel", "ezra"]},
],
}
plan = generate_plan(spec, {"fleet": {"ezra": "ONLINE", "bezalel": "ONLINE"}})
assert plan["assignments"] == [
{"task_id": "ISSUE-9", "host": "bezalel", "lane": "general", "priority": 100}
]
def test_generate_plan_reports_unassigned_when_no_host_matches_lane():
spec = {
"hosts": [
{"name": "ezra", "capacity": 1, "lanes": ["research"]},
],
"tasks": [
{"id": "ISSUE-5", "lane": "build", "priority": 50},
],
}
plan = generate_plan(spec, {"fleet": {"ezra": "ONLINE"}})
assert plan["assignments"] == []
assert plan["unassigned"] == [
{"task_id": "ISSUE-5", "reason": "no_online_host_for_lane:build"}
]
def test_write_plan_persists_json(tmp_path):
plan = {
"assignments": [{"task_id": "ISSUE-1", "host": "ezra", "lane": "general", "priority": 10}],
"offline_hosts": [],
"unassigned": [],
}
output_path = tmp_path / "dispatch-plan.json"
write_plan(plan, output_path)
assert json.loads(output_path.read_text()) == plan