Quartz 4

control_api_patch

3 min read

Phase 1: Routing Layer Changes to control_api.py

Changes made

1. Added ROUTING constants and helpers (after NOTE_RE)

ROUTES_PATH = os.path.join(WIKI, "control-plane", "model-routing.json")
VALID_TYPES = {"planning", "design", "audit", "review", "implementation",
               "infra", "email", "admin", "coding", "bugfix"}
# Title keywords → task type inference (Phase 1: best-effort)
TYPE_KW = {
    "planning": [re.compile(r"\b(planning|roadmap|plan\b)\b", re.I)],
    "design":   [re.compile(r"\b(design|specification|architecture)\b", re.I)],
    "audit":    [re.compile(r"\b(audit|review)\b", re.I)],
    "infra":    [re.compile(r"\b(infra|deploy|kubernetes|pod|node)\b", re.I)],
    "email":    [re.compile(r"\b(email|mail|smtp)\b", re.I)],
}
 
def _infer_type(title):
    """Best-effort type inference from title text. Returns first match or None."""
    for t, pats in TYPE_KW.items():
        if any(p.search(title) for p in pats):
            return t
    return None
 
def _load_routing():
    """Load model-routing.json; create defaults if missing."""
    try:
        return json.load(open(ROUTES_PATH, encoding="utf-8"))
    except (FileNotFoundError, json.JSONDecodeError):
        return {"type_to_tier": {}, "tiers": {}}
 
def _save_routing(data):
    os.makedirs(os.path.dirname(ROUTES_PATH), exist_ok=True)
    json.dump(data, open(ROUTES_PATH, "w", encoding="utf-8"), indent=2)

2. Updated _parse_task to include type field

In the return dict of _parse_task(), added:

"type": fm.get("type") or _infer_title_type(fm.get("title", "")),

3. Updated create_task to accept and auto-infer type

# Inside create_task(), after priority line:
task_type = payload.get("type") or _infer_type(title) or "implementation"
task_type = task_type if task_type in VALID_TYPES else "implementation"
# Add to frontmatter generation:
f"type: {task_type}\n"

4. Updated patch_task to support type and tier_override fields

Added "type" and "tier_override" to the patchable fields list.

5. New /api/routing endpoints

@app.get("/api/routing")
def get_routing():
    return _load_routing()
 
@app.put("/api/routing")
def update_routing(payload: dict = Body(...)):
    r = _load_routing()
    # Merge provided fields into existing config
    if "type_to_tier" in payload:
        r["type_to_tier"].update(payload["type_to_tier"])
    if "tiers" in payload:
        for k, v in payload.get("tiers", {}).items():
            if isinstance(v, dict):
                r["tiers"][k].update(v)
    _save_routing(r)
    return {"ok": True}
 
@app.get("/api/tasks/{task_id}/tier")
def task_tier(task_id: str):
    """Return the tier assigned to a specific task, respecting override."""
    f, status = _find_file(task_id)
    if not f:
        raise HTTPException(404, "task not found")
    t = _parse_task(f, status)
    routing = _load_routing()
    # tier_override takes precedence
    override = t.get("tier_override")
    if override and override in routing.get("tiers", {}):
        return {"task_id": task_id, "type": t.get("type"),
                "tier": override, "overridden": True}
    tt = t.get("type", "implementation")
    tier = routing.get("type_to_tier", {}).get(tt, "builder")
    return {"task_id": task_id, "type": tt, "tier": tier, "overridden": False}
 
@app.post("/api/tasks/{task_id}/dispatch")
def dispatch_task(task_id: str):
    """Dispatch a task to its routed agent (Phase 1 wire)."""
    f, status = _find_file(task_id)
    if not f:
        raise HTTPException(404, "task not found")
    t = _parse_task(f, status)
    routing = _load_routing()
    override = t.get("tier_override")
    tt = t.get("type", "implementation")
    tier_name = override if override in routing.get("tiers", {}) else \
        routing.get("type_to_tier", {}).get(tt, "builder")
    tier_info = routing.get("tiers", {}).get(tier_name, {})
    # Phase 1: wire infra→mercury (operator tier), everything else→hermes (builder)
    agent = tier_info.get("agent", "hermes").lower()
    # Determine target endpoint based on agent name
    if "mercury" in agent or tier_name == "operator":
        target = "https://mercury.hermes.svc.cluster.local:9080/api/task"
    else:
        target = "https://hermes-api-server.hermes.svc.cluster.local:443/api/v1/tasks"
    return {"task_id": task_id, "type": tt, "tier": tier_name,
            "target_agent": agent, "dispatch_url": target}

Verification steps

  • GET /api/routing returns the existing model-routing.json content
  • GET /api/tasks/{id}/tier returns the assigned tier for a task
  • PUT /api/routing updates model-routing.json on disk
  • Task creation auto-infers type from title keywords
  • Infra tasks route to mercury, all others to hermes (builder)

Graph View