ive harnessed the harness
Something went wrong. Try again.
3.0 kB · 76 lines
Python
at main
1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677"""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()