iop/agent-ops/skills/project/orchestrate-agent-task-loop/scripts/dispatcher_observation.py

21 lines
617 B
Python

#!/usr/bin/env python3
"""Observation output emitter and formatting utilities for agent-task dispatcher."""
from __future__ import annotations
SEP = "-" * 42
def banner(event: str, task: str, lines: list[str] | None = None) -> None:
display_task = task.rsplit("/", 1)[-1]
print(SEP, flush=True)
print(f"{event}: {display_task}", flush=True)
print(SEP, flush=True)
if display_task != task:
print(f"task={task}", flush=True)
for line in lines or []:
print(line, flush=True)
def attempt_event(prefix: str, message: str) -> None:
print(f"{prefix} {message}", flush=True)