diff --git a/api/main.py b/api/main.py index 8f277ff..92fbc79 100644 --- a/api/main.py +++ b/api/main.py @@ -17,6 +17,7 @@ from api.routers import interface_changes from api.routers import flows from api.routers import recently_on_scope from api.routers import reconciliation +from api.routers import execution class ETagMiddleware(BaseHTTPMiddleware): @@ -102,6 +103,7 @@ app.include_router(token_events.router) app.include_router(interface_changes.router) app.include_router(flows.router) app.include_router(reconciliation.router) +app.include_router(execution.router) app.include_router(state.router) app.include_router(policy.router) diff --git a/api/models/__init__.py b/api/models/__init__.py index c3377d0..5bf0fb7 100644 --- a/api/models/__init__.py +++ b/api/models/__init__.py @@ -21,6 +21,7 @@ from api.models.tpsc import TPSCCatalog, TPSCSnapshot, TPSCEntry from api.models.doi_cache import DOICache from api.models.token_event import TokenEvent from api.models.interface_change import InterfaceChange +from api.models.workplan_launch_request import WorkplanLaunchRequest __all__ = [ "Base", @@ -46,4 +47,5 @@ __all__ = [ "DOICache", "TokenEvent", "InterfaceChange", + "WorkplanLaunchRequest", ] diff --git a/api/models/workplan_launch_request.py b/api/models/workplan_launch_request.py new file mode 100644 index 0000000..72b770e --- /dev/null +++ b/api/models/workplan_launch_request.py @@ -0,0 +1,39 @@ +import uuid + +from sqlalchemy import Boolean, ForeignKey, String, Text +from sqlalchemy.dialects.postgresql import JSONB, UUID +from sqlalchemy.orm import Mapped, mapped_column, relationship + +from api.models.base import Base, TimestampMixin, new_uuid + + +class WorkplanLaunchRequest(Base, TimestampMixin): + __tablename__ = "workplan_launch_requests" + + id: Mapped[uuid.UUID] = mapped_column( + UUID(as_uuid=True), primary_key=True, default=new_uuid + ) + workstream_id: Mapped[uuid.UUID] = mapped_column( + UUID(as_uuid=True), + ForeignKey("workstreams.id", ondelete="CASCADE"), + nullable=False, + index=True, + ) + requested_by: Mapped[str] = mapped_column(String(100), nullable=False, default="dashboard") + requested_actor: Mapped[str | None] = mapped_column(String(100), nullable=True) + launch_mode: Mapped[str] = mapped_column(String(20), nullable=False, default="queued", index=True) + concurrency_mode: Mapped[str] = mapped_column(String(20), nullable=False, default="sequential", index=True) + priority: Mapped[str | None] = mapped_column(String(20), nullable=True, index=True) + repo_id: Mapped[uuid.UUID | None] = mapped_column( + UUID(as_uuid=True), + ForeignKey("managed_repos.id", ondelete="SET NULL"), + nullable=True, + index=True, + ) + branch_preference: Mapped[str | None] = mapped_column(Text, nullable=True) + immediate_pickup: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False, server_default="false") + status: Mapped[str] = mapped_column(String(20), nullable=False, default="requested", server_default="requested", index=True) + notes: Mapped[str | None] = mapped_column(Text, nullable=True) + request_metadata: Mapped[dict] = mapped_column(JSONB, nullable=False, default=dict, server_default="{}") + + workstream: Mapped["Workstream"] = relationship("Workstream", back_populates="launch_requests") # noqa: F821 diff --git a/api/models/workstream.py b/api/models/workstream.py index 9b3223b..bf38be1 100644 --- a/api/models/workstream.py +++ b/api/models/workstream.py @@ -1,7 +1,7 @@ import uuid -from datetime import date +from datetime import date, datetime -from sqlalchemy import Date, ForeignKey, Integer, String, Text +from sqlalchemy import Date, DateTime, ForeignKey, Integer, String, Text from sqlalchemy.dialects.postgresql import UUID from sqlalchemy.orm import Mapped, mapped_column, relationship @@ -27,6 +27,18 @@ class Workstream(Base, TimestampMixin): due_date: Mapped[date | None] = mapped_column(Date, nullable=True) planning_priority: Mapped[str | None] = mapped_column(String(20), nullable=True, index=True) planning_order: Mapped[int | None] = mapped_column(Integer, nullable=True, index=True) + execution_state: Mapped[str] = mapped_column( + String(20), nullable=False, default="manual", server_default="manual", index=True + ) + launch_mode: Mapped[str] = mapped_column( + String(20), nullable=False, default="manual", server_default="manual", index=True + ) + concurrency_mode: Mapped[str] = mapped_column( + String(20), nullable=False, default="sequential", server_default="sequential", index=True + ) + queue_rank: Mapped[int | None] = mapped_column(Integer, nullable=True, index=True) + execution_group: Mapped[str | None] = mapped_column(String(100), nullable=True, index=True) + scheduled_for: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True, index=True) repo_id: Mapped[uuid.UUID | None] = mapped_column( UUID(as_uuid=True), @@ -53,3 +65,6 @@ class Workstream(Base, TimestampMixin): progress_events: Mapped[list["ProgressEvent"]] = relationship( # noqa: F821 "ProgressEvent", back_populates="workstream", lazy="selectin" ) + launch_requests: Mapped[list["WorkplanLaunchRequest"]] = relationship( # noqa: F821 + "WorkplanLaunchRequest", back_populates="workstream", lazy="selectin" + ) diff --git a/api/routers/execution.py b/api/routers/execution.py new file mode 100644 index 0000000..28bb363 --- /dev/null +++ b/api/routers/execution.py @@ -0,0 +1,196 @@ +import uuid + +from fastapi import APIRouter, Depends, HTTPException, Query, status +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession + +from api.database import get_session +from api.models.task import Task, TaskStatus +from api.models.workplan_launch_request import WorkplanLaunchRequest +from api.models.workstream import Workstream +from api.models.workstream_dependency import WorkstreamDependency +from api.schemas.execution import ( + ExecutionIntentRead, + ExecutionIntentUpdate, + ExecutionSemantics, + LaunchRequestCreate, + LaunchRequestRead, + WorkplanQueueItem, +) +from api.services.execution_queue import ( + ACTIVITY_CORE_RESPONSIBILITIES, + CONCURRENCY_MODES, + EXECUTION_STATES, + LAUNCH_MODES, + STATE_HUB_RESPONSIBILITIES, + execution_state_for_launch, + queue_sort_key, + workstream_blockers, +) +from api.workplan_status import CLOSED_WORKSTREAM_STATUSES, normalize_workstream_status + +router = APIRouter(prefix="/execution", tags=["execution"]) + + +@router.get("/semantics", response_model=ExecutionSemantics) +async def execution_semantics() -> ExecutionSemantics: + return ExecutionSemantics( + execution_states=EXECUTION_STATES, + launch_modes=LAUNCH_MODES, + concurrency_modes=CONCURRENCY_MODES, + state_hub_responsibility=STATE_HUB_RESPONSIBILITIES, + activity_core_responsibility=ACTIVITY_CORE_RESPONSIBILITIES, + ) + + +@router.patch("/workstreams/{workstream_id}/intent", response_model=ExecutionIntentRead) +async def update_execution_intent( + workstream_id: uuid.UUID, + body: ExecutionIntentUpdate, + session: AsyncSession = Depends(get_session), +) -> ExecutionIntentRead: + ws = await session.get(Workstream, workstream_id) + if ws is None: + raise HTTPException(status_code=404, detail="Workstream not found") + + for field, value in body.model_dump(exclude_unset=True).items(): + setattr(ws, field, value) + await session.commit() + await session.refresh(ws) + return _intent_read(ws) + + +@router.get("/workplan-stack", response_model=list[WorkplanQueueItem]) +async def workplan_stack( + include_manual: bool = Query(True), + include_blocked: bool = Query(True), + session: AsyncSession = Depends(get_session), +) -> list[WorkplanQueueItem]: + result = await session.execute(select(Workstream)) + workstreams = [ + ws for ws in result.scalars().all() + if normalize_workstream_status(ws.status) not in CLOSED_WORKSTREAM_STATUSES + ] + ws_by_id = {ws.id: ws for ws in workstreams} + ws_status = {ws.id: normalize_workstream_status(ws.status) for ws in workstreams} + + dep_result = await session.execute(select(WorkstreamDependency)) + ws_deps: dict[uuid.UUID, list[uuid.UUID]] = {} + task_deps: dict[uuid.UUID, list[uuid.UUID]] = {} + for dep in dep_result.scalars().all(): + if dep.to_workstream_id is not None: + ws_deps.setdefault(dep.from_workstream_id, []).append(dep.to_workstream_id) + if dep.to_task_id is not None: + task_deps.setdefault(dep.from_workstream_id, []).append(dep.to_task_id) + + task_ids = [task_id for ids in task_deps.values() for task_id in ids] + task_status: dict[uuid.UUID, str] = {} + if task_ids: + task_result = await session.execute(select(Task).where(Task.id.in_(task_ids))) + task_status = {task.id: _task_status(task.status) for task in task_result.scalars().all()} + + items: list[WorkplanQueueItem] = [] + for ws in workstreams: + if not include_manual and ws.execution_state == "manual": + continue + lifecycle_status = normalize_workstream_status(ws.status) + blocked_ws = [ + blocker for blocker in workstream_blockers(ws.id, ws_deps, ws_status) + if blocker in ws_by_id or blocker in ws_status + ] + blocked_tasks = [ + task_id for task_id in task_deps.get(ws.id, []) + if task_status.get(task_id) not in {"done", "cancelled"} + ] + eligible = lifecycle_status != "blocked" and not blocked_ws and not blocked_tasks + if not include_blocked and not eligible: + continue + sort_key = queue_sort_key(ws, eligible=eligible) + items.append(WorkplanQueueItem( + workstream_id=ws.id, + slug=ws.slug, + title=ws.title, + status=lifecycle_status, + repo_id=ws.repo_id, + planning_priority=ws.planning_priority, + planning_order=ws.planning_order, + execution_state=ws.execution_state, + launch_mode=ws.launch_mode, + concurrency_mode=ws.concurrency_mode, + queue_rank=ws.queue_rank, + execution_group=ws.execution_group, + scheduled_for=ws.scheduled_for, + eligible=eligible, + blocked_by_workstream_ids=blocked_ws, + blocked_by_task_ids=blocked_tasks, + sort_key=sort_key, + )) + return sorted(items, key=lambda item: item.sort_key) + + +@router.post( + "/launch-requests", + response_model=LaunchRequestRead, + status_code=status.HTTP_201_CREATED, +) +async def create_launch_request( + body: LaunchRequestCreate, + session: AsyncSession = Depends(get_session), +) -> WorkplanLaunchRequest: + ws = await session.get(Workstream, body.workstream_id) + if ws is None: + raise HTTPException(status_code=404, detail="Workstream not found") + + launch_request = WorkplanLaunchRequest( + workstream_id=ws.id, + requested_by=body.requested_by, + requested_actor=body.requested_actor, + launch_mode=body.launch_mode, + concurrency_mode=body.concurrency_mode, + priority=body.priority or ws.planning_priority, + repo_id=body.repo_id or ws.repo_id, + branch_preference=body.branch_preference, + immediate_pickup=body.immediate_pickup, + notes=body.notes, + request_metadata=body.request_metadata, + ) + ws.launch_mode = body.launch_mode + ws.concurrency_mode = body.concurrency_mode + ws.execution_state = execution_state_for_launch(body.launch_mode, body.immediate_pickup) + session.add(launch_request) + await session.commit() + await session.refresh(launch_request) + return launch_request + + +@router.get("/launch-requests", response_model=list[LaunchRequestRead]) +async def list_launch_requests( + workstream_id: uuid.UUID | None = None, + request_status: str | None = None, + session: AsyncSession = Depends(get_session), +) -> list[WorkplanLaunchRequest]: + q = select(WorkplanLaunchRequest).order_by(WorkplanLaunchRequest.created_at.desc()) + if workstream_id: + q = q.where(WorkplanLaunchRequest.workstream_id == workstream_id) + if request_status: + q = q.where(WorkplanLaunchRequest.status == request_status) + result = await session.execute(q) + return list(result.scalars().all()) + + +def _intent_read(ws: Workstream) -> ExecutionIntentRead: + return ExecutionIntentRead( + workstream_id=ws.id, + execution_state=ws.execution_state, + launch_mode=ws.launch_mode, + concurrency_mode=ws.concurrency_mode, + queue_rank=ws.queue_rank, + execution_group=ws.execution_group, + scheduled_for=ws.scheduled_for, + ) + + +def _task_status(status_value: TaskStatus | str) -> str: + if hasattr(status_value, "value"): + return status_value.value + return str(status_value or "").strip().lower() diff --git a/api/schemas/execution.py b/api/schemas/execution.py new file mode 100644 index 0000000..60799e6 --- /dev/null +++ b/api/schemas/execution.py @@ -0,0 +1,91 @@ +import uuid +from datetime import datetime +from typing import Literal + +from pydantic import BaseModel, ConfigDict, Field + + +ExecutionState = Literal["manual", "queued", "scheduled", "launching", "paused", "completed", "cancelled"] +LaunchMode = Literal["manual", "queued", "scheduled", "immediate"] +ConcurrencyMode = Literal["sequential", "parallel"] +LaunchRequestStatus = Literal["requested", "accepted", "completed", "cancelled"] + + +class ExecutionIntentUpdate(BaseModel): + execution_state: ExecutionState | None = None + launch_mode: LaunchMode | None = None + concurrency_mode: ConcurrencyMode | None = None + queue_rank: int | None = None + execution_group: str | None = None + scheduled_for: datetime | None = None + + +class ExecutionIntentRead(BaseModel): + workstream_id: uuid.UUID + execution_state: ExecutionState + launch_mode: LaunchMode + concurrency_mode: ConcurrencyMode + queue_rank: int | None = None + execution_group: str | None = None + scheduled_for: datetime | None = None + + +class WorkplanQueueItem(BaseModel): + workstream_id: uuid.UUID + slug: str + title: str + status: str + repo_id: uuid.UUID | None = None + planning_priority: str | None = None + planning_order: int | None = None + execution_state: ExecutionState + launch_mode: LaunchMode + concurrency_mode: ConcurrencyMode + queue_rank: int | None = None + execution_group: str | None = None + scheduled_for: datetime | None = None + eligible: bool + blocked_by_workstream_ids: list[uuid.UUID] = Field(default_factory=list) + blocked_by_task_ids: list[uuid.UUID] = Field(default_factory=list) + sort_key: list[str | int] = Field(default_factory=list) + + +class LaunchRequestCreate(BaseModel): + workstream_id: uuid.UUID + requested_by: str = "dashboard" + requested_actor: str | None = None + launch_mode: LaunchMode = "queued" + concurrency_mode: ConcurrencyMode = "sequential" + priority: str | None = None + repo_id: uuid.UUID | None = None + branch_preference: str | None = None + immediate_pickup: bool = False + notes: str | None = None + request_metadata: dict = Field(default_factory=dict) + + +class LaunchRequestRead(BaseModel): + model_config = ConfigDict(from_attributes=True) + id: uuid.UUID + workstream_id: uuid.UUID + requested_by: str + requested_actor: str | None = None + launch_mode: LaunchMode + concurrency_mode: ConcurrencyMode + priority: str | None = None + repo_id: uuid.UUID | None = None + branch_preference: str | None = None + immediate_pickup: bool + status: LaunchRequestStatus + notes: str | None = None + request_metadata: dict = Field(default_factory=dict) + created_at: datetime + updated_at: datetime + + +class ExecutionSemantics(BaseModel): + execution_states: dict[str, str] + launch_modes: dict[str, str] + concurrency_modes: dict[str, str] + state_hub_responsibility: list[str] + activity_core_responsibility: list[str] diff --git a/api/schemas/workstream.py b/api/schemas/workstream.py index 29fb614..b7e8a0f 100644 --- a/api/schemas/workstream.py +++ b/api/schemas/workstream.py @@ -16,6 +16,9 @@ WorkstreamStatus = Literal[ "finished", "archived", ] +ExecutionState = Literal["manual", "queued", "scheduled", "launching", "paused", "completed", "cancelled"] +LaunchMode = Literal["manual", "queued", "scheduled", "immediate"] +ConcurrencyMode = Literal["sequential", "parallel"] class WorkstreamStatusMixin(BaseModel): @@ -35,6 +38,12 @@ class WorkstreamCreate(WorkstreamStatusMixin): due_date: date | None = None planning_priority: str | None = None planning_order: int | None = None + execution_state: ExecutionState = "manual" + launch_mode: LaunchMode = "manual" + concurrency_mode: ConcurrencyMode = "sequential" + queue_rank: int | None = None + execution_group: str | None = None + scheduled_for: datetime | None = None repo_id: uuid.UUID | None = None # GEMS primary: the owning repository repo_goal_id: uuid.UUID | None = None @@ -47,6 +56,12 @@ class WorkstreamUpdate(WorkstreamStatusMixin): due_date: date | None = None planning_priority: str | None = None planning_order: int | None = None + execution_state: ExecutionState | None = None + launch_mode: LaunchMode | None = None + concurrency_mode: ConcurrencyMode | None = None + queue_rank: int | None = None + execution_group: str | None = None + scheduled_for: datetime | None = None repo_id: uuid.UUID | None = None repo_goal_id: uuid.UUID | None = None @@ -65,6 +80,12 @@ class WorkstreamRead(WorkstreamStatusMixin): due_date: date | None = None planning_priority: str | None = None planning_order: int | None = None + execution_state: ExecutionState = "manual" + launch_mode: LaunchMode = "manual" + concurrency_mode: ConcurrencyMode = "sequential" + queue_rank: int | None = None + execution_group: str | None = None + scheduled_for: datetime | None = None created_at: datetime updated_at: datetime diff --git a/api/services/execution_queue.py b/api/services/execution_queue.py new file mode 100644 index 0000000..07f86ff --- /dev/null +++ b/api/services/execution_queue.py @@ -0,0 +1,97 @@ +from __future__ import annotations + +from typing import Any + +from api.workplan_status import normalize_workstream_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": "A launch request asks for immediate pickup or has been handed off.", + "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": "Request prompt activity-core or agent pickup.", +} + +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", + "record launch requests and handoff metadata durably", + "surface manual, queued, scheduled, and immediate intent to operators", +] + +ACTIVITY_CORE_RESPONSIBILITIES = [ + "own schedules, wakeups, and recurring automation", + "dispatch coding agents and coordinate parallel execution", + "acknowledge, run, and complete launch requests when available", +] + +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_WORKSTREAM_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 workstream_blockers( + workstream_id: Any, + dependency_targets: dict[Any, list[Any]], + workstream_status: dict[Any, str], +) -> list[Any]: + blockers = [] + for target_id in dependency_targets.get(workstream_id, []): + target_status = normalize_workstream_status(workstream_status.get(target_id)) + if target_status not in CLOSED_WORKSTREAM_STATUSES: + blockers.append(target_id) + return 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 ""), + ] diff --git a/dashboard/src/workplan-queue.md b/dashboard/src/workplan-queue.md new file mode 100644 index 0000000..b3af3cf --- /dev/null +++ b/dashboard/src/workplan-queue.md @@ -0,0 +1,291 @@ +--- +title: Workplan Queue +--- + +```js +import {apiFetch, waitForVisible, pollDelay, POLL_HEAVY} from "./components/config.js"; +``` + +```js +const queueState = (async function*() { + let failures = 0; + while (true) { + let stack = [], semantics = {}, ok = false; + try { + const [stackResponse, semanticsResponse] = await Promise.all([ + apiFetch("/execution/workplan-stack"), + apiFetch("/execution/semantics"), + ]); + ok = stackResponse.ok && semanticsResponse.ok; + if (ok) { + [stack, semantics] = await Promise.all([ + stackResponse.json(), + semanticsResponse.json(), + ]); + } + } catch {} + failures = ok ? 0 : failures + 1; + yield {stack, semantics, ok, ts: new Date()}; + await waitForVisible(pollDelay({ok, base: POLL_HEAVY, failures})); + } +})(); +``` + +```js +const stack = queueState.stack ?? []; +const semantics = queueState.semantics ?? {}; +const _ok = queueState.ok ?? false; +const _ts = queueState.ts; +``` + +# Workplan Queue + +```js +display(html`
+ ● + ${_ok ? `Live · updated ${_ts?.toLocaleTimeString()}` : html`Offline`} +
`); +``` + +```js +const launchModes = Object.keys(semantics.launch_modes ?? {manual: "", queued: "", scheduled: "", immediate: ""}); +const concurrencyModes = Object.keys(semantics.concurrency_modes ?? {sequential: "", parallel: ""}); + +function optionList(values, selected) { + return values.map(value => html``); +} + +function statusCell(row) { + const classes = ["queue-status", row.eligible ? "eligible" : "blocked"].join(" "); + return html`${row.eligible ? "eligible" : "blocked"}`; +} + +function blockers(row) { + const parts = []; + if (row.blocked_by_workstream_ids?.length) parts.push(`${row.blocked_by_workstream_ids.length} workstream`); + if (row.blocked_by_task_ids?.length) parts.push(`${row.blocked_by_task_ids.length} task`); + return parts.length ? parts.join(", ") : "—"; +} + +function queueControls(row) { + const root = html`
`; + const mode = html``; + const concurrency = html``; + const rank = html``; + const group = html``; + const message = html``; + const save = html``; + const launch = html``; + + const payload = () => ({ + execution_state: mode.value === "manual" ? "manual" : mode.value === "scheduled" ? "scheduled" : "queued", + launch_mode: mode.value, + concurrency_mode: concurrency.value, + queue_rank: rank.value === "" ? null : Number(rank.value), + execution_group: group.value.trim() || null, + }); + + async function run(label, action) { + message.textContent = label; + message.className = "queue-message"; + save.disabled = true; + launch.disabled = true; + try { + await action(); + message.textContent = "saved"; + message.classList.add("ok"); + setTimeout(() => location.reload(), 450); + } catch (error) { + message.textContent = error?.message ?? "failed"; + message.classList.add("error"); + save.disabled = false; + launch.disabled = false; + } + } + + save.onclick = () => run("saving", async () => { + const response = await apiFetch(`/execution/workstreams/${row.workstream_id}/intent`, { + method: "PATCH", + headers: {"Content-Type": "application/json"}, + body: JSON.stringify(payload()), + }); + if (!response.ok) throw new Error(`HTTP ${response.status}`); + }); + + launch.onclick = () => run("requesting", async () => { + const intent = payload(); + const response = await apiFetch("/execution/launch-requests", { + method: "POST", + headers: {"Content-Type": "application/json"}, + body: JSON.stringify({ + workstream_id: row.workstream_id, + requested_by: "dashboard", + requested_actor: "activity-core", + launch_mode: intent.launch_mode, + concurrency_mode: intent.concurrency_mode, + immediate_pickup: intent.launch_mode === "immediate", + priority: row.planning_priority, + notes: `Queue request from dashboard for ${row.slug}`, + }), + }); + if (!response.ok) throw new Error(`HTTP ${response.status}`); + }); + + root.append(mode, concurrency, rank, group, save, launch, message); + return root; +} +``` + +```js +if (stack.length === 0) { + display(html`

No queue candidates.

`); +} else { + display(html` + + + + + + + + + + + + + ${stack.map(row => html` + + + + + + + + + `)} +
StateRankWorkplanLifecyclePriorityEligibilityBlocked ByIntent
${row.execution_state}${row.queue_rank ?? row.planning_order ?? "—"}${row.slug}
${row.title}
${row.status}${row.planning_priority ?? "—"}${statusCell(row)}${blockers(row)}${queueControls(row)}
`); +} +``` + + diff --git a/dashboard/src/workstreams.md b/dashboard/src/workstreams.md index 1b866a4..73286ee 100644 --- a/dashboard/src/workstreams.md +++ b/dashboard/src/workstreams.md @@ -215,6 +215,7 @@ injectTocTop("live-indicator", _liveEl); const _h1 = document.querySelector("#observablehq-main h1"); if (_h1) { _h1.style.position = "relative"; withDocHelp(_h1, "/docs/workstreams"); } +display(html`

Workplan queue

`); ``` ```js diff --git a/migrations/versions/x1s2t3u4v5w6_workplan_execution_queue.py b/migrations/versions/x1s2t3u4v5w6_workplan_execution_queue.py new file mode 100644 index 0000000..147627c --- /dev/null +++ b/migrations/versions/x1s2t3u4v5w6_workplan_execution_queue.py @@ -0,0 +1,89 @@ +"""add workplan execution queue and launch requests + +Revision ID: x1s2t3u4v5w6 +Revises: w0r1s2t3u4v5 +Create Date: 2026-05-23 + +""" +from alembic import op +import sqlalchemy as sa +from sqlalchemy.dialects.postgresql import JSONB, UUID + +revision = "x1s2t3u4v5w6" +down_revision = "w0r1s2t3u4v5" +branch_labels = None +depends_on = None + + +def upgrade() -> None: + op.add_column( + "workstreams", + sa.Column("execution_state", sa.String(length=20), nullable=False, server_default="manual"), + ) + op.add_column( + "workstreams", + sa.Column("launch_mode", sa.String(length=20), nullable=False, server_default="manual"), + ) + op.add_column( + "workstreams", + sa.Column("concurrency_mode", sa.String(length=20), nullable=False, server_default="sequential"), + ) + op.add_column("workstreams", sa.Column("queue_rank", sa.Integer(), nullable=True)) + op.add_column("workstreams", sa.Column("execution_group", sa.String(length=100), nullable=True)) + op.add_column("workstreams", sa.Column("scheduled_for", sa.DateTime(timezone=True), nullable=True)) + op.create_index("ix_workstreams_execution_state", "workstreams", ["execution_state"]) + op.create_index("ix_workstreams_launch_mode", "workstreams", ["launch_mode"]) + op.create_index("ix_workstreams_concurrency_mode", "workstreams", ["concurrency_mode"]) + op.create_index("ix_workstreams_queue_rank", "workstreams", ["queue_rank"]) + op.create_index("ix_workstreams_execution_group", "workstreams", ["execution_group"]) + op.create_index("ix_workstreams_scheduled_for", "workstreams", ["scheduled_for"]) + + op.create_table( + "workplan_launch_requests", + sa.Column("id", UUID(as_uuid=True), primary_key=True, nullable=False), + sa.Column("workstream_id", UUID(as_uuid=True), nullable=False), + sa.Column("requested_by", sa.String(length=100), nullable=False), + sa.Column("requested_actor", sa.String(length=100), nullable=True), + sa.Column("launch_mode", sa.String(length=20), nullable=False), + sa.Column("concurrency_mode", sa.String(length=20), nullable=False), + sa.Column("priority", sa.String(length=20), nullable=True), + sa.Column("repo_id", UUID(as_uuid=True), nullable=True), + sa.Column("branch_preference", sa.Text(), nullable=True), + sa.Column("immediate_pickup", sa.Boolean(), nullable=False, server_default="false"), + sa.Column("status", sa.String(length=20), nullable=False, server_default="requested"), + sa.Column("notes", sa.Text(), nullable=True), + sa.Column("request_metadata", JSONB(), nullable=False, server_default="{}"), + sa.Column("created_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now()), + sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now()), + sa.ForeignKeyConstraint(["repo_id"], ["managed_repos.id"], ondelete="SET NULL"), + sa.ForeignKeyConstraint(["workstream_id"], ["workstreams.id"], ondelete="CASCADE"), + ) + op.create_index("ix_workplan_launch_requests_workstream_id", "workplan_launch_requests", ["workstream_id"]) + op.create_index("ix_workplan_launch_requests_launch_mode", "workplan_launch_requests", ["launch_mode"]) + op.create_index("ix_workplan_launch_requests_concurrency_mode", "workplan_launch_requests", ["concurrency_mode"]) + op.create_index("ix_workplan_launch_requests_priority", "workplan_launch_requests", ["priority"]) + op.create_index("ix_workplan_launch_requests_repo_id", "workplan_launch_requests", ["repo_id"]) + op.create_index("ix_workplan_launch_requests_status", "workplan_launch_requests", ["status"]) + + +def downgrade() -> None: + op.drop_index("ix_workplan_launch_requests_status", table_name="workplan_launch_requests") + op.drop_index("ix_workplan_launch_requests_repo_id", table_name="workplan_launch_requests") + op.drop_index("ix_workplan_launch_requests_priority", table_name="workplan_launch_requests") + op.drop_index("ix_workplan_launch_requests_concurrency_mode", table_name="workplan_launch_requests") + op.drop_index("ix_workplan_launch_requests_launch_mode", table_name="workplan_launch_requests") + op.drop_index("ix_workplan_launch_requests_workstream_id", table_name="workplan_launch_requests") + op.drop_table("workplan_launch_requests") + + op.drop_index("ix_workstreams_scheduled_for", table_name="workstreams") + op.drop_index("ix_workstreams_execution_group", table_name="workstreams") + op.drop_index("ix_workstreams_queue_rank", table_name="workstreams") + op.drop_index("ix_workstreams_concurrency_mode", table_name="workstreams") + op.drop_index("ix_workstreams_launch_mode", table_name="workstreams") + op.drop_index("ix_workstreams_execution_state", table_name="workstreams") + op.drop_column("workstreams", "scheduled_for") + op.drop_column("workstreams", "execution_group") + op.drop_column("workstreams", "queue_rank") + op.drop_column("workstreams", "concurrency_mode") + op.drop_column("workstreams", "launch_mode") + op.drop_column("workstreams", "execution_state") diff --git a/tests/test_routers_core.py b/tests/test_routers_core.py index e41d40a..494cb7b 100644 --- a/tests/test_routers_core.py +++ b/tests/test_routers_core.py @@ -806,3 +806,139 @@ class TestReconciliationEndpoints: r = await client.get(f"/tasks/{task['id']}") assert r.json()["status"] == "todo" + + +class TestExecutionQueueEndpoints: + async def test_execution_semantics_separates_state_hub_and_activity_core(self, client): + r = await client.get("/execution/semantics") + + assert r.status_code == 200 + body = r.json() + assert "queued" in body["execution_states"] + assert "immediate" in body["launch_modes"] + assert "parallel" in body["concurrency_modes"] + assert any("launch requests" in item for item in body["state_hub_responsibility"]) + assert any("dispatch" in item for item in body["activity_core_responsibility"]) + + async def test_execution_intent_update_does_not_change_lifecycle_status(self, client): + await _create_domain(client) + topic = await _create_topic(client) + ws = await _create_workstream(client, topic["id"], status="ready") + + r = await client.patch(f"/execution/workstreams/{ws['id']}/intent", json={ + "execution_state": "queued", + "launch_mode": "queued", + "concurrency_mode": "parallel", + "queue_rank": 7, + "execution_group": "ui-state", + }) + + assert r.status_code == 200, r.text + body = r.json() + assert body["execution_state"] == "queued" + assert body["launch_mode"] == "queued" + assert body["concurrency_mode"] == "parallel" + assert body["queue_rank"] == 7 + + r = await client.get(f"/workstreams/{ws['id']}") + assert r.json()["status"] == "ready" + assert r.json()["execution_state"] == "queued" + + async def test_workplan_stack_orders_eligible_queued_work_before_blocked(self, client): + await _create_domain(client) + topic = await _create_topic(client) + queued = await _create_workstream( + client, + topic["id"], + slug="queued-wp", + status="ready", + planning_priority="high", + planning_order=2, + ) + blocked = await _create_workstream( + client, + topic["id"], + slug="blocked-wp", + status="ready", + planning_priority="high", + planning_order=1, + ) + lifecycle_blocked = await _create_workstream( + client, + topic["id"], + slug="lifecycle-blocked-wp", + status="blocked", + planning_priority="high", + planning_order=0, + ) + dependency = await _create_workstream( + client, + topic["id"], + slug="dependency-wp", + status="active", + planning_priority="low", + planning_order=3, + ) + await client.patch(f"/execution/workstreams/{queued['id']}/intent", json={ + "execution_state": "queued", + "launch_mode": "queued", + "queue_rank": 2, + }) + await client.patch(f"/execution/workstreams/{blocked['id']}/intent", json={ + "execution_state": "queued", + "launch_mode": "queued", + "queue_rank": 1, + }) + await client.patch(f"/execution/workstreams/{lifecycle_blocked['id']}/intent", json={ + "execution_state": "queued", + "launch_mode": "queued", + "queue_rank": 0, + }) + await client.post( + f"/workstreams/{blocked['id']}/dependencies/", + json={"to_workstream_id": dependency["id"], "description": "wait"}, + ) + + r = await client.get("/execution/workplan-stack?include_manual=false") + + assert r.status_code == 200, r.text + rows = r.json() + assert [row["slug"] for row in rows] == ["queued-wp", "lifecycle-blocked-wp", "blocked-wp"] + assert rows[0]["eligible"] is True + assert rows[1]["eligible"] is False + assert rows[1]["status"] == "blocked" + assert rows[2]["eligible"] is False + assert rows[2]["blocked_by_workstream_ids"] == [dependency["id"]] + + async def test_launch_request_records_handoff_and_updates_execution_intent(self, client): + await _create_domain(client) + topic = await _create_topic(client) + ws = await _create_workstream(client, topic["id"], status="ready") + + r = await client.post("/execution/launch-requests", json={ + "workstream_id": ws["id"], + "requested_by": "dashboard", + "requested_actor": "activity-core", + "launch_mode": "immediate", + "concurrency_mode": "sequential", + "priority": "high", + "branch_preference": "codex/state-wp-0049", + "immediate_pickup": True, + "notes": "start now", + }) + + assert r.status_code == 201, r.text + body = r.json() + assert body["workstream_id"] == ws["id"] + assert body["launch_mode"] == "immediate" + assert body["immediate_pickup"] is True + assert body["status"] == "requested" + + r = await client.get(f"/workstreams/{ws['id']}") + updated = r.json() + assert updated["status"] == "ready" + assert updated["execution_state"] == "launching" + assert updated["launch_mode"] == "immediate" + + r = await client.get(f"/execution/launch-requests?workstream_id={ws['id']}") + assert len(r.json()) == 1 diff --git a/workplans/STATE-WP-0049-workplan-execution-queue.md b/workplans/STATE-WP-0049-workplan-execution-queue.md index b1455ac..cbc10b9 100644 --- a/workplans/STATE-WP-0049-workplan-execution-queue.md +++ b/workplans/STATE-WP-0049-workplan-execution-queue.md @@ -4,7 +4,7 @@ type: workplan title: "Workplan Execution Queue and Activity-Core Handoff" domain: custodian repo: state-hub -status: proposed +status: finished owner: codex topic_slug: custodian planning_priority: medium @@ -38,7 +38,7 @@ automations, and agent dispatch. ```task id: STATE-WP-0049-T01 -status: todo +status: done priority: high state_hub_task_id: "b69e3c7d-46be-4e88-b039-7aa82d653b53" ``` @@ -49,11 +49,17 @@ sequential, and parallel intent. Done when queue semantics are documented without overloading lifecycle status. +Result 2026-05-23: added execution semantics for `manual`, `queued`, +`scheduled`, `launching`, `paused`, `completed`, and `cancelled` execution +states; `manual`, `queued`, `scheduled`, and `immediate` launch modes; and +`sequential`/`parallel` concurrency modes. The `/execution/semantics` endpoint +exposes the vocabulary and boundary notes. + ## T02 - Separate Lifecycle From Execution Intent ```task id: STATE-WP-0049-T02 -status: todo +status: done priority: high state_hub_task_id: "4f48b4d4-7232-4e36-b4a2-03aaef684ab4" ``` @@ -63,11 +69,16 @@ execution intent lives in separate planning/execution fields. Done when `active` no longer has to imply "dispatch a coding agent right now." +Result 2026-05-23: added separate execution intent fields on workstreams: +`execution_state`, `launch_mode`, `concurrency_mode`, `queue_rank`, +`execution_group`, and `scheduled_for`. Lifecycle `status` remains unchanged +when execution intent is updated. + ## T03 - Add Prioritized Workplan Stack ```task id: STATE-WP-0049-T03 -status: todo +status: done priority: high state_hub_task_id: "9237b778-4c83-4ebe-942e-79bae29aaa45" ``` @@ -78,11 +89,15 @@ optional concurrency grouping. Done when a human can choose what starts next from a clear ordered list. +Result 2026-05-23: added `/execution/workplan-stack`, a dependency-aware +ordered queue view using execution state, planning priority, queue rank, +planning order, lifecycle status, and workstream/task blockers. + ## T04 - Add Launch Request Records ```task id: STATE-WP-0049-T04 -status: todo +status: done priority: medium state_hub_task_id: "a8366495-fd66-436e-b423-e24c3ffadbc9" ``` @@ -94,11 +109,16 @@ preference, and whether immediate agent pickup is desired. Done when UI/API actions can request implementation without losing intent in a plain status patch. +Result 2026-05-23: added durable `workplan_launch_requests` records and +`POST /execution/launch-requests`. Launch requests capture requested actor, +mode, priority, repo, branch preference, immediate pickup, notes, and metadata +while updating execution intent separately from lifecycle status. + ## T05 - Define Activity-Core Boundary ```task id: STATE-WP-0049-T05 -status: todo +status: done priority: high state_hub_task_id: "2ebf5186-9a92-476a-ac62-3588cc51fb31" ``` @@ -110,11 +130,15 @@ own scheduling, wakeups, and agent dispatch when available. Done when the boundary is explicit enough that implementation can proceed in State Hub now without trapping orchestration there permanently. +Result 2026-05-23: the execution semantics service defines State Hub as the +state, ranking, and durable handoff owner, while activity-core owns schedules, +wakeups, agent dispatch, and request execution once available. + ## T06 - Dashboard Controls ```task id: STATE-WP-0049-T06 -status: todo +status: done priority: medium state_hub_task_id: "7d1c9ec7-8130-4f26-92ae-754c6f28ea48" ``` @@ -125,11 +149,15 @@ intent, and manual versus immediate pickup. Done when a user can move a workplan into the queue without accidentally starting work, or explicitly request immediate pickup when desired. +Result 2026-05-23: added the dashboard Workplan Queue page with queue rank, +launch mode, sequential/parallel intent, execution group, save, and request +controls. Workstreams link to the queue page. + ## T07 - Handoff And Scheduling Tests ```task id: STATE-WP-0049-T07 -status: todo +status: done priority: medium state_hub_task_id: "f819907b-0409-411e-82d1-564c2d543f86" ``` @@ -139,6 +167,10 @@ eligibility, and the State Hub/activity-core boundary contract. Done when execution intent can evolve without contaminating lifecycle state. +Result 2026-05-23: added API tests for semantics, execution-intent updates, +dependency-aware queue ordering, and launch-request creation. The tests pin +that lifecycle status remains separate from execution intent. + ## Acceptance Criteria - Workplan lifecycle status is separate from execution intent.