mirror of
https://github.com/razzant/ouroboros.git
synced 2026-10-03 12:18:39 +00:00
f3ed795ea1
Keep the same task and live browser while required owner input lends active worker capacity. Restore acknowledged planned continuations without replaying effects or changing the original budget threshold. Align visible Publish admission, root questions, ongoing replies, display fallback names and definite publication failures with existing owners. Co-authored-by: Ouroboros <311266734+ouroboros-agent@users.noreply.github.com>
77 lines
3.8 KiB
Python
77 lines
3.8 KiB
Python
"""Supervisor projection of live model waits; the queue still owns task lifetime."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import copy
|
|
from typing import Any
|
|
from ouroboros.model_wait import quota_waited_seconds # noqa: F401 -- historical supervisor import surface
|
|
|
|
|
|
def model_waiting(meta: dict) -> bool:
|
|
"""A typed wait spares idle detection; auth still spends execution time."""
|
|
task = meta.get("task") or {}
|
|
attempt = int(meta.get("attempt") or task.get("_attempt") or 1)
|
|
return any(isinstance(row, dict) and row.get("state") == "waiting"
|
|
and row.get("task_attempt") == attempt
|
|
for row in (task.get("model_waits") or {}).values())
|
|
|
|
|
|
def handle_task_model_wait(event: dict, ctx: Any) -> None:
|
|
"""Apply a current-attempt, monotonic worker fact without minting a task."""
|
|
from supervisor.queue import _queue_lock
|
|
from supervisor.log_addressing import address_ctx_event
|
|
|
|
task_id, wait_id = str(event.get("task_id") or ""), str(event.get("wait_id") or "")
|
|
revision, attempt = event.get("revision"), event.get("task_attempt")
|
|
if (not task_id or not wait_id or not isinstance(revision, int) or isinstance(revision, bool)
|
|
or revision <= 0 or not isinstance(attempt, int) or isinstance(attempt, bool)
|
|
or event.get("state") not in {"waiting", "resolved"}):
|
|
return
|
|
payload = copy.deepcopy(event)
|
|
row = {key: value for key, value in payload.items()
|
|
if key not in {"type", "ts", "task_id", "quota_clock", "is_progress"}}
|
|
with _queue_lock:
|
|
meta = ctx.RUNNING.get(task_id)
|
|
if isinstance(meta, dict):
|
|
task = meta.get("task") or {}
|
|
expected = int(meta.get("attempt") or task.get("_attempt") or 1)
|
|
if attempt != expected:
|
|
return
|
|
waits = dict(task.get("model_waits") or {})
|
|
previous = waits.get(wait_id) or {}
|
|
if int(previous.get("revision") or 0) >= revision:
|
|
return
|
|
waits[wait_id] = row
|
|
task["model_waits"] = waits
|
|
previous_clock = meta.get("model_wait_quota_clock") or {}
|
|
clock = payload.get("quota_clock") or {}
|
|
if int(clock.get("revision") or 0) > int(previous_clock.get("revision") or 0):
|
|
meta["model_wait_quota_clock"] = clock
|
|
else:
|
|
# A finished managed task cannot be resurrected by a late waiter.
|
|
# Direct turns have no queue row and retain their own live owner.
|
|
from supervisor.workers import direct_chat_turn
|
|
|
|
if direct_chat_turn(task_id) is None:
|
|
from supervisor.active_activity import get_direct_activity_registry
|
|
|
|
consciousness = getattr(ctx, "consciousness", None)
|
|
owner = get_direct_activity_registry().ephemeral_model_wait(ctx.DRIVE_ROOT, task_id)
|
|
if owner is None and consciousness and task_id == "bg-consciousness":
|
|
owner = consciousness.live_model_wait()
|
|
if owner is None:
|
|
from ouroboros.post_task_checkpoint import post_task_model_wait
|
|
owner = post_task_model_wait(ctx.DRIVE_ROOT, task_id)
|
|
if owner is None or payload.get("model_wait_owner_id", "") != owner.owner_id:
|
|
return
|
|
with owner.lock:
|
|
current = owner.waits.get(wait_id) or {}
|
|
if owner.closed or attempt != owner.attempt or current.get("revision") != revision:
|
|
return
|
|
payload["chat_id"] = consciousness._owner_chat_id_fn() if task_id == "bg-consciousness" else owner.task.get("chat_id")
|
|
address_ctx_event(ctx, payload)
|
|
ctx.append_jsonl(ctx.DRIVE_ROOT / "logs" / "progress.jsonl", payload)
|
|
ctx.bridge.push_log(payload)
|
|
|
|
|
|
EVENT_HANDLERS = {"task_model_wait": handle_task_model_wait}
|