feat: migrate missing harnesses, templates, and workflows from agent-runtimes
Brings the framework CRS repo up to date with all content that was living in agent-runtimes (local-dev fallback) but hadn't been promoted. New composites: feature-delivery-loop, integration-direct, scaffolding-repo, sonnet-impl-narrow, sonnet-manager, test-writing-repo New contexts: integration/v1, scaffolding/v1, sonnet-manager/v1, z-ai/v1, airouter/v1/bin (anthropic-compat-wrapper.sh), cp-harness/v1/init.sh New task-templates: sonnet-integrator.yaml, workflow/* (17 typed workflow task templates for the Epic 1 pipeline) Updated: agent-repo/v1/finalize.sh — adds AR-38/F97 empty-deliverable audit (SKIP_BRANCH_PUSH support, boilerplate-path filtering, ci_metadata.json flag) Also adds MEMORY.md index and memory/ topic files for the framework repo.
This commit is contained in:
Paul O'Reilly
301
harnesses/contexts/sonnet-manager/v1/runner.py
Normal file
301
harnesses/contexts/sonnet-manager/v1/runner.py
Normal file
@@ -0,0 +1,301 @@
|
||||
"""Sonnet-manager runner — per-project orchestrator.
|
||||
|
||||
Drives planning items toward terminal states by polling the ACL inbox,
|
||||
checking eligibility, and dispatching workflows. Idle-exits when both
|
||||
inbox and eligibility have been empty for settle_threshold consecutive
|
||||
polls.
|
||||
|
||||
Implements MS-18..MS-25 from spec/manager-sonnet.md.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import sys
|
||||
import time
|
||||
from dataclasses import dataclass
|
||||
from subprocess import CalledProcessError
|
||||
|
||||
__all__ = [
|
||||
"IdleExitConfig",
|
||||
"manager_loop",
|
||||
"handle_inbox_message",
|
||||
"fallback_escalation",
|
||||
]
|
||||
|
||||
_logger = logging.getLogger("manager.runner")
|
||||
|
||||
|
||||
# ── Data model ───────────────────────────────────────────────────────────────
|
||||
|
||||
@dataclass
|
||||
class IdleExitConfig:
|
||||
"""Idle-exit configuration.
|
||||
|
||||
settle_threshold: number of consecutive empty iterations before exiting.
|
||||
poll_interval_seconds: sleep between iterations when idle.
|
||||
"""
|
||||
settle_threshold: int
|
||||
poll_interval_seconds: float = 30.0
|
||||
|
||||
|
||||
# ── Main loop (MS-18, MS-19, MS-20, MS-24, MS-25) ──────────────────────────
|
||||
|
||||
def manager_loop(*, client, project_id, idle_config: IdleExitConfig):
|
||||
"""Outer loop: inbox → eligibility → dispatch → idle-exit.
|
||||
|
||||
MS-18: inbox polled BEFORE eligibility each iteration.
|
||||
MS-19: exit 0 after settle_threshold consecutive empty iterations.
|
||||
MS-20: tag_revoke before sys.exit(0).
|
||||
MS-24: revoke → final poll → re-advertise-if-nonempty.
|
||||
MS-25: acl_send exit 3 triggers fallback, not crash.
|
||||
MS-9: tag_advertise exit 3 is fatal.
|
||||
"""
|
||||
# Startup: advertise orchestrator tag
|
||||
try:
|
||||
client.tag_advertise(project_id=project_id)
|
||||
except CalledProcessError as exc:
|
||||
# MS-9: tag_advertise exit 3 is fatal
|
||||
_logger.error("tag_advertise failed (exit %s) — fatal", exc.returncode)
|
||||
sys.exit(exc.returncode if exc.returncode else 1)
|
||||
|
||||
idle_settle_count = 0
|
||||
|
||||
while True:
|
||||
# ── MS-18: inbox poll FIRST ──────────────────────────────────────
|
||||
inbox = client.acl_inbox_poll(project_id=project_id)
|
||||
|
||||
if inbox:
|
||||
# Non-empty inbox → reset settle counter
|
||||
idle_settle_count = 0
|
||||
# Process each message
|
||||
for msg in inbox:
|
||||
_process_inbox_message_safe(client, msg, project_id)
|
||||
# After processing, go back to top of loop
|
||||
continue
|
||||
|
||||
# ── Eligibility check ─────────────────────────────────────────────
|
||||
eligible = client.list_eligible_workflows(project_id=project_id)
|
||||
|
||||
if not eligible:
|
||||
idle_settle_count += 1
|
||||
else:
|
||||
idle_settle_count = 0
|
||||
# Per-item dispatch would go here (MS-14..MS-16, not yet implemented)
|
||||
# for item in eligible:
|
||||
# rule = policy.match(item, item.get('eligible', []))
|
||||
# if rule:
|
||||
# client.dispatch_workflow(...)
|
||||
|
||||
if idle_settle_count >= idle_config.settle_threshold:
|
||||
# ── MS-20 / MS-24: idle-exit sequence ─────────────────────────
|
||||
# MS-20: revoke before exit
|
||||
client.tag_revoke(project_id=project_id)
|
||||
|
||||
# MS-24: final poll after revoke
|
||||
final_inbox = client.acl_inbox_poll(project_id=project_id)
|
||||
if final_inbox:
|
||||
# Re-advertise and reset counter
|
||||
client.tag_advertise(project_id=project_id)
|
||||
idle_settle_count = 0
|
||||
continue
|
||||
|
||||
# Final poll empty → clean exit
|
||||
sys.exit(0)
|
||||
|
||||
if idle_config.poll_interval_seconds > 0:
|
||||
time.sleep(idle_config.poll_interval_seconds)
|
||||
|
||||
|
||||
def _process_inbox_message_safe(client, message, project_id):
|
||||
"""Process a single inbox message, catching exceptions.
|
||||
|
||||
MS-25: acl_send exit 3 should not crash the manager.
|
||||
"""
|
||||
message_type = message.get("message_type")
|
||||
if message_type is None:
|
||||
# Plain message — log and continue
|
||||
_logger.info("Received plain message (msg_id=%s): %s",
|
||||
message.get("msg_id"), message.get("body", ""))
|
||||
return
|
||||
|
||||
try:
|
||||
# We need item_uuid from the message; fall back to None
|
||||
item_uuid = message.get("item_uuid")
|
||||
handle_inbox_message(
|
||||
client=client,
|
||||
message=message,
|
||||
item_uuid=item_uuid,
|
||||
project_id=project_id,
|
||||
)
|
||||
except CalledProcessError as exc:
|
||||
if exc.returncode == 3:
|
||||
# MS-25: exit 3 from acl_send → fallback escalation
|
||||
payload = message.get("typed_payload") or {}
|
||||
fallback_escalation(
|
||||
client=client,
|
||||
original_message_type=message_type,
|
||||
cp_error_code="typed_message_unroutable",
|
||||
typed_payload=payload,
|
||||
item_uuid=item_uuid,
|
||||
project_id=project_id,
|
||||
)
|
||||
else:
|
||||
_logger.error("acl_send failed with exit %s: %s", exc.returncode, exc)
|
||||
|
||||
|
||||
# ── Inbox message handler (MS-21, MS-22) ─────────────────────────────────────
|
||||
|
||||
def handle_inbox_message(*, client, message, item_uuid, project_id):
|
||||
"""Route a typed inbox message according to its message_type.
|
||||
|
||||
MS-21: request-handoff → escalation map dispatch or human fallback.
|
||||
MS-22: request-clarification → policy-driven routing.
|
||||
MS-25: acl_send exit 3 → fallback_escalation (caught by caller).
|
||||
"""
|
||||
msg_type = message.get("message_type")
|
||||
typed_payload = message.get("typed_payload") or {}
|
||||
|
||||
if msg_type == "request-handoff":
|
||||
_handle_request_handoff(client, message, typed_payload, item_uuid, project_id)
|
||||
elif msg_type == "request-clarification":
|
||||
_handle_request_clarification(client, message, typed_payload, item_uuid, project_id)
|
||||
else:
|
||||
_logger.info("Unhandled message type '%s' (msg_id=%s)", msg_type, message.get("msg_id"))
|
||||
|
||||
|
||||
def _handle_request_handoff(client, message, typed_payload, item_uuid, project_id):
|
||||
"""MS-21: dispatch escalation workflow or fall back to human."""
|
||||
reason = typed_payload.get("reason", "other")
|
||||
|
||||
# "other" always goes to human
|
||||
if reason == "other":
|
||||
_escalate_to_human(client, message, typed_payload, item_uuid, project_id)
|
||||
return
|
||||
|
||||
# Look up escalation map from workflow template
|
||||
try:
|
||||
template = client.get_workflow_template(item_uuid=item_uuid, project_id=project_id)
|
||||
except Exception:
|
||||
template = {}
|
||||
|
||||
escalation_map = template.get("escalation") or {}
|
||||
target_workflow = escalation_map.get(reason)
|
||||
|
||||
if target_workflow:
|
||||
# MS-21: dispatch configured escalation workflow
|
||||
try:
|
||||
client.dispatch_workflow(
|
||||
item_uuid=item_uuid,
|
||||
workflow=target_workflow,
|
||||
project_id=project_id,
|
||||
)
|
||||
except CalledProcessError as exc:
|
||||
if exc.returncode == 3:
|
||||
fallback_escalation(
|
||||
client=client,
|
||||
original_message_type="request-handoff",
|
||||
cp_error_code=getattr(exc, "cp_error_code", "dispatch_failed"),
|
||||
typed_payload=typed_payload,
|
||||
item_uuid=item_uuid,
|
||||
project_id=project_id,
|
||||
)
|
||||
else:
|
||||
raise
|
||||
else:
|
||||
# No escalation for this reason → human fallback
|
||||
_escalate_to_human(client, message, typed_payload, item_uuid, project_id)
|
||||
|
||||
|
||||
def _handle_request_clarification(client, message, typed_payload, item_uuid, project_id):
|
||||
"""MS-22: request-clarification is routed to human for policy decision."""
|
||||
_escalate_to_human(client, message, typed_payload, item_uuid, project_id)
|
||||
|
||||
|
||||
def _escalate_to_human(client, message, typed_payload, item_uuid, project_id):
|
||||
"""Send a request-clarification to tag: human."""
|
||||
try:
|
||||
client.acl_send(
|
||||
to="tag:human",
|
||||
message_type="request-clarification",
|
||||
body=_build_escalation_body(message, typed_payload),
|
||||
item_uuid=item_uuid,
|
||||
project_id=project_id,
|
||||
)
|
||||
except CalledProcessError as exc:
|
||||
if exc.returncode == 3:
|
||||
fallback_escalation(
|
||||
client=client,
|
||||
original_message_type="request-clarification",
|
||||
cp_error_code=getattr(exc, "cp_error_code", "acl_send_failed"),
|
||||
typed_payload=typed_payload,
|
||||
item_uuid=item_uuid,
|
||||
project_id=project_id,
|
||||
)
|
||||
else:
|
||||
raise
|
||||
|
||||
|
||||
def _build_escalation_body(message, typed_payload):
|
||||
"""Build a human-readable escalation body from a message."""
|
||||
context_ref = typed_payload.get("context_ref")
|
||||
context_digest = ""
|
||||
if context_ref:
|
||||
context_digest = f" Context: {repr(context_ref)}"
|
||||
return (
|
||||
f"Handoff/clarification request: reason={typed_payload.get('reason', 'unknown')}."
|
||||
f"{context_digest}"
|
||||
)
|
||||
|
||||
|
||||
# ── Fallback escalation (MS-25) ──────────────────────────────────────────────
|
||||
|
||||
def fallback_escalation(*, client, original_message_type, cp_error_code,
|
||||
typed_payload, item_uuid, project_id):
|
||||
"""MS-25: when a typed send fails, send a plain escalation to human.
|
||||
|
||||
If the fallback send also fails, log a warning and continue.
|
||||
The manager MUST NOT crash.
|
||||
"""
|
||||
_logger.warning(
|
||||
"fallback escalation: typed send failed — type=%s, error=%s",
|
||||
original_message_type, cp_error_code,
|
||||
)
|
||||
|
||||
body = _build_fallback_body(original_message_type, cp_error_code, typed_payload)
|
||||
|
||||
try:
|
||||
client.acl_send(
|
||||
to="tag:human",
|
||||
body=body,
|
||||
item_uuid=item_uuid,
|
||||
project_id=project_id,
|
||||
)
|
||||
_logger.info(
|
||||
"fallback escalation sent successfully for item %s", item_uuid,
|
||||
)
|
||||
except CalledProcessError as exc:
|
||||
_logger.warning(
|
||||
"fallback_escalation_failed: original_type=%s, cp_error=%s, "
|
||||
"fallback_exit=%s",
|
||||
original_message_type, cp_error_code, exc.returncode,
|
||||
)
|
||||
|
||||
|
||||
def _build_fallback_body(original_message_type, cp_error_code, typed_payload):
|
||||
"""Build the fallback plain-message body (MS-25 format).
|
||||
|
||||
Includes original message type, CP error code, truncated payload summary,
|
||||
and a SERIALIZER_VERSION drift note.
|
||||
"""
|
||||
reason = typed_payload.get("reason", "unknown") if isinstance(typed_payload, dict) else "unknown"
|
||||
# Truncate payload summary to 500 chars
|
||||
payload_summary = repr(typed_payload)[:500]
|
||||
|
||||
body = (
|
||||
f"Manager could not route typed message: type={original_message_type}, "
|
||||
f"reason={reason}.\n"
|
||||
f"CP rejected with: {cp_error_code}. "
|
||||
f"Possible cause: SERIALIZER_VERSION drift (AC-43).\n"
|
||||
f"Payload summary: {payload_summary}"
|
||||
)
|
||||
return body
|
||||
Reference in New Issue
Block a user