state-hub/api/services/execution_queue.py

from __future__ import annotations

from typing import Any

from api.workplan_status import normalize_workplan_status


EXECUTION_STATES = {
    "manual": "Not queued for autonomous pickup; humans or agents may still work manually.",
    "queued": "Candidate for ordered pickup when dependencies and concurrency allow it.",
    "scheduled": "Waiting for an external launch window; State Hub stores the requested time.",
    "launching": "Legacy state only; it does not prove that any consumer accepted pickup.",
    "paused": "Temporarily held outside the pickup stack.",
    "completed": "Execution intent is closed; lifecycle status remains authoritative.",
    "cancelled": "Execution intent was cancelled without changing lifecycle status.",
}

LAUNCH_MODES = {
    "manual": "Do not request automation; keep intent visible only.",
    "queued": "Place in the prioritized stack for later pickup.",
    "scheduled": "Request pickup at or after a selected time.",
    "immediate": "Legacy intent only; workplan launch-request pickup is retired.",
}

CONCURRENCY_MODES = {
    "sequential": "Respect queue order and avoid parallel pickup for the same group.",
    "parallel": "Eligible for concurrent pickup with other ready work.",
}

STATE_HUB_RESPONSIBILITIES = [
    "store lifecycle status separately from execution intent",
    "rank candidate workplans and expose dependency-aware eligibility",
    "preserve historical launch-request and execution-intent rows during retirement",
    "reject new workplan launch requests because no pickup consumer exists",
]

ACTIVITY_CORE_RESPONSIBILITIES = [
    "own schedules, wakeups, and recurring automation",
    "claim and complete ops runs created by ActivityDefinition fires",
    "do not consume State Hub workplan launch requests",
]

EXECUTION_REPLACEMENTS = {
    "POST /execution/launch-requests": (
        "Queue work in the authoritative repository file; for recurring or operational "
        "fires use an ActivityDefinition and activity-core ops_run"
    ),
    "GET /execution/launch-requests": "GET /ops-runs on activity-core for automation history",
    "GET /execution/workplan-stack": "Repo Manager work index or hub-core projection",
    "PATCH /execution/workplans/{id}/intent": "Edit the authoritative workplan file",
}

EXECUTION_STATE_RANK = {
    "launching": 0,
    "queued": 1,
    "scheduled": 2,
    "manual": 3,
    "paused": 4,
    "completed": 5,
    "cancelled": 6,
}

PRIORITY_RANK = {
    "critical": 0,
    "high": 1,
    "medium": 2,
    "low": 3,
}

CLOSED_WORKPLAN_STATUSES = {"finished", "archived"}


def execution_state_for_launch(launch_mode: str, immediate_pickup: bool = False) -> str:
    mode = (launch_mode or "queued").strip().lower()
    if immediate_pickup or mode == "immediate":
        return "launching"
    if mode == "scheduled":
        return "scheduled"
    if mode == "manual":
        return "manual"
    return "queued"


def workplan_blockers(
    workplan_id: Any,
    dependency_targets: dict[Any, list[Any]],
    workplan_status: dict[Any, str],
    workstream_id: Any = None,
) -> list[Any]:
    scope_id = workplan_id if workplan_id is not None else workstream_id
    blockers = []
    for target_id in dependency_targets.get(scope_id, []):
        target_status = normalize_workplan_status(workplan_status.get(target_id))
        if target_status not in CLOSED_WORKPLAN_STATUSES:
            blockers.append(target_id)
    return blockers


workstream_blockers = workplan_blockers


def queue_sort_key(workstream: Any, *, eligible: bool) -> list[int | str]:
    priority = str(getattr(workstream, "planning_priority", "") or "").strip().lower()
    execution_state = str(getattr(workstream, "execution_state", "") or "manual").strip().lower()
    return [
        0 if eligible else 1,
        EXECUTION_STATE_RANK.get(execution_state, 99),
        PRIORITY_RANK.get(priority, 50),
        getattr(workstream, "queue_rank", None) if getattr(workstream, "queue_rank", None) is not None else 999_999,
        getattr(workstream, "planning_order", None) if getattr(workstream, "planning_order", None) is not None else 999_999,
        str(getattr(workstream, "slug", "") or ""),
    ]