Allow channel overrides during escalation

humanlayer-ts/src/models.ts

@@ -87,6 +87,7 @@ type FunctionCall = {
 type Escalation = {
   escalation_msg: string
   additional_recipients?: EmailRecipient[]
+  channel?: ContactChannel
 }
 
 type HumanContactSpec = {

humanlayer-ts/src/models_escalation.test.ts

@@ -0,0 +1,53 @@
+import { ContactChannel, EmailContactChannel, Escalation } from './models'
+
+test('Escalation type - basic functionality', () => {
+  // Test escalation without channel
+  const escalation: Escalation = {
+    escalation_msg: 'Test escalation',
+  }
+  expect(escalation.channel).toBeUndefined()
+  expect(escalation.escalation_msg).toBe('Test escalation')
+  expect(escalation.additional_recipients).toBeUndefined()
+})
+
+test('Escalation type - with channel', () => {
+  // Test escalation with channel
+  const emailChannel: EmailContactChannel = {
+    address: 'test@example.com',
+  }
+  const contactChannel: ContactChannel = {
+    email: emailChannel,
+  }
+  const escalationWithChannel: Escalation = {
+    escalation_msg: 'Escalation with channel',
+    channel: contactChannel,
+  }
+
+  expect(escalationWithChannel.channel).toBeDefined()
+  expect(escalationWithChannel.channel?.email).toBeDefined()
+  expect(escalationWithChannel.channel?.email?.address).toBe('test@example.com')
+  expect(escalationWithChannel.escalation_msg).toBe('Escalation with channel')
+})
+
+test('Escalation type - serialization includes channel', () => {
+  // Test escalation serializes channel field correctly
+  const emailChannel: EmailContactChannel = {
+    address: 'ceo@company.com',
+    experimental_subject_line: 'CRITICAL: Immediate approval required',
+  }
+  const contactChannel: ContactChannel = {
+    email: emailChannel,
+  }
+  const escalation: Escalation = {
+    escalation_msg: 'CRITICAL: Still no response',
+    channel: contactChannel,
+  }
+
+  // Test JSON serialization includes channel field
+  const serialized = JSON.parse(JSON.stringify(escalation))
+  expect(serialized.channel).toBeDefined()
+  expect(serialized.channel.email.address).toBe('ceo@company.com')
+  expect(serialized.channel.email.experimental_subject_line).toBe(
+    'CRITICAL: Immediate approval required'
+  )
+})

humanlayer/core/models.py

@@ -233,6 +233,7 @@ class FunctionCall(BaseModel):
 class Escalation(BaseModel):
     escalation_msg: str
     additional_recipients: list[EmailRecipient] | None = None
+    channel: ContactChannel | None = None
 
 
 class HumanContactSpec(BaseModel):

humanlayer/core/models_test.py

@@ -1,7 +1,7 @@
 import pytest
 from pydantic import ValidationError
 
-from humanlayer.core.models import SlackContactChannel
+from humanlayer.core.models import ContactChannel, EmailContactChannel, Escalation, SlackContactChannel
 
 
 def test_slack_contact_channel_allows_none_and_nonempty_list() -> None:
@@ -24,3 +24,35 @@ def test_slack_contact_channel_rejects_empty_list() -> None:
 def test_slack_contact_channel_allows_none() -> None:
     channel = SlackContactChannel(channel_or_user_id="C123", allowed_responder_ids=None)
     assert channel.allowed_responder_ids is None
+
+
+def test_escalation_channel_field() -> None:
+    # Test escalation without channel
+    escalation = Escalation(escalation_msg="Test escalation")
+    assert escalation.channel is None
+    assert escalation.escalation_msg == "Test escalation"
+    assert escalation.additional_recipients is None
+
+    # Test escalation with channel
+    email_channel = EmailContactChannel(address="test@example.com")
+    contact_channel = ContactChannel(email=email_channel)
+    escalation_with_channel = Escalation(escalation_msg="Escalation with channel", channel=contact_channel)
+    assert escalation_with_channel.channel is not None
+    assert escalation_with_channel.channel.email is not None
+    assert escalation_with_channel.channel.email.address == "test@example.com"
+    assert escalation_with_channel.escalation_msg == "Escalation with channel"
+
+
+def test_escalation_serialization() -> None:
+    # Test escalation serializes channel field correctly
+    email_channel = EmailContactChannel(
+        address="ceo@company.com", experimental_subject_line="CRITICAL: Immediate approval required"
+    )
+    contact_channel = ContactChannel(email=email_channel)
+    escalation = Escalation(escalation_msg="CRITICAL: Still no response", channel=contact_channel)
+
+    # Test model_dump includes channel field
+    dumped = escalation.model_dump()
+    assert "channel" in dumped
+    assert dumped["channel"]["email"]["address"] == "ceo@company.com"
+    assert dumped["channel"]["email"]["experimental_subject_line"] == "CRITICAL: Immediate approval required"