"""Compatibility defaults: optional ambient participation, deliberate speech, and agent-operated focus."""
from klbr.hooks import (
AttentionDecision,
AttentionPosture,
AttentionRequest,
DeliveryDecision,
DeliveryRequest,
HookContext,
get_posture,
record_held_conversation,
select_available_posture,
)
def is_diagnostic_text(text: str) -> bool:
"""Identifies diagnostic traces, check outputs, or tool execution errors that must remain internal."""
lower = text.lower()
if "traceback (most recent call last):" in lower:
return True
if "[diagnostic]" in lower or "checkresult:" in lower:
return True
if lower.startswith("error:") or lower.startswith("fatal:") or lower.startswith("panic:"):
return True
if "diagnosticclaim" in lower or "maintenanceassessment" in lower:
return True
return False
def attention(request: AttentionRequest, context: HookContext) -> AttentionDecision:
# Origin invariance: origin swap alone (human vs agent) does not change trust/priority
# (Contract G05, SWARM §10). Priority is derived strictly from source and urgency.
posture = get_posture()
# If posture has an expiry and deadline has passed, revert to available
if posture.is_focused and posture.is_expired(request.received_ms):
posture = select_available_posture()
if posture.is_focused:
# Check agreed bypasses (operator, urgent dependencies, whitelisted authors/channels)
if posture.should_bypass(request):
if request.source == "operator":
return AttentionDecision.wake("operator")
else:
return AttentionDecision.wake("directed")
# Non-bypassing traffic during focus is held in durable backlog, not discarded
record_held_conversation(
request,
reason=f"held by {posture.name} posture on {posture.target or 'work'}",
)
return AttentionDecision.defer(60)
# Available posture (standard defaults)
match request.source:
case "operator":
return AttentionDecision.wake("operator")
case "discord.dm" | "discord.mention":
return AttentionDecision.wake("directed")
case "urgent_dependency":
return AttentionDecision.wake("directed")
case "discord.ambient":
return AttentionDecision.defer(8)
if request.is_urgent:
return AttentionDecision.wake("directed")
raise ValueError(f"unknown source {request.source!r}; never silently discard it")
def delivery(request: DeliveryRequest, context: HookContext) -> DeliveryDecision:
# Diagnostics must remain internal: tool errors, checkup traces, and stack traces
# do not become speech directed to operator.
if is_diagnostic_text(request.text):
return DeliveryDecision.hold("diagnostics do not become speech")
return DeliveryDecision.allow()