f5968cfab5
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.
302 lines
11 KiB
Python
302 lines
11 KiB
Python
"""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
|