mirror of
https://github.com/dograh-hq/dograh.git
synced 2026-07-22 11:51:04 +02:00
fix: save call metadata in gathered context for an api trigger outbou… (#566)
* fix: save call metadata in gathered context for an api trigger outbound call * fix: handle workflow run update failures * fix: format * fix: pin integration workflow runs to published definitions * test: expect text chat runtime configuration
This commit is contained in:
Sabiha Khan
GitHub
parent
84d3cd0c03
commit
ded23b1daa
5 changed files with 133 additions and 2 deletions
|
|
@ -320,7 +320,7 @@ async def _execute_resolved_target(
|
|||
# (e.g. Telnyx, Cloudonix); without them the URL contains "None/None" and
|
||||
# the stream connection fails.
|
||||
try:
|
||||
await provider.initiate_call(
|
||||
result = await provider.initiate_call(
|
||||
to_number=request.phone_number,
|
||||
webhook_url=webhook_url,
|
||||
workflow_run_id=workflow_run.id,
|
||||
|
|
@ -337,6 +337,25 @@ async def _execute_resolved_target(
|
|||
detail=f"Failed to initiate call: {e}",
|
||||
)
|
||||
|
||||
gathered_context = {
|
||||
"provider": provider.PROVIDER_NAME,
|
||||
"triggered_by": "api",
|
||||
**(result.provider_metadata or {}),
|
||||
}
|
||||
if target.identifier_type == "trigger_path":
|
||||
gathered_context["trigger_uuid"] = target.identifier_value
|
||||
|
||||
try:
|
||||
await db_client.update_workflow_run(
|
||||
run_id=workflow_run.id,
|
||||
gathered_context=gathered_context,
|
||||
)
|
||||
except Exception as e:
|
||||
logger.warning(
|
||||
f"Call initiated for workflow run {workflow_run.id}, but failed to "
|
||||
f"persist provider metadata: {e}"
|
||||
)
|
||||
|
||||
logger.info(
|
||||
f"Call initiated successfully for workflow run {workflow_run.id} "
|
||||
f"via {target.identifier_type}={target.identifier_value}"
|
||||
|
|
|
|||
|
|
@ -1332,6 +1332,7 @@ class XAITTSConfiguration(BaseServiceConfiguration):
|
|||
description="BCP-47 language code for synthesis (e.g. 'en', 'fr', 'de'), or 'auto' for automatic language detection.",
|
||||
json_schema_extra={"allow_custom_input": True},
|
||||
)
|
||||
|
||||
@computed_field
|
||||
@property
|
||||
def model(self) -> str:
|
||||
|
|
|
|||
|
|
@ -236,6 +236,7 @@ async def create_workflow_run_rows(
|
|||
workflow_id=workflow.id,
|
||||
mode=WorkflowRunMode.SMALLWEBRTC.value,
|
||||
user_id=user.id,
|
||||
definition_id=workflow.released_definition_id,
|
||||
)
|
||||
|
||||
return workflow_run, user, workflow
|
||||
|
|
|
|||
|
|
@ -45,7 +45,14 @@ def _provider():
|
|||
PROVIDER_NAME="twilio",
|
||||
WEBHOOK_ENDPOINT="outbound",
|
||||
validate_config=Mock(return_value=True),
|
||||
initiate_call=AsyncMock(),
|
||||
initiate_call=AsyncMock(
|
||||
return_value=SimpleNamespace(
|
||||
call_id="CA123",
|
||||
status="queued",
|
||||
caller_number="+15550000000",
|
||||
provider_metadata={"call_id": "CA123"},
|
||||
)
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
|
|
@ -96,6 +103,7 @@ def test_trigger_route_executes_as_workflow_owner():
|
|||
return_value=SimpleNamespace(id=55)
|
||||
)
|
||||
mock_db.create_workflow_run = AsyncMock(return_value=SimpleNamespace(id=501))
|
||||
mock_db.update_workflow_run = AsyncMock()
|
||||
|
||||
response = client.post(
|
||||
"/public/agent/trigger-uuid-123",
|
||||
|
|
@ -135,6 +143,15 @@ def test_trigger_route_executes_as_workflow_owner():
|
|||
assert initiate_kwargs["workflow_id"] == workflow.id
|
||||
# The media websocket URL is keyed on the org, not the workflow owner.
|
||||
assert initiate_kwargs["organization_id"] == workflow.organization_id
|
||||
mock_db.update_workflow_run.assert_awaited_once_with(
|
||||
run_id=501,
|
||||
gathered_context={
|
||||
"provider": "twilio",
|
||||
"triggered_by": "api",
|
||||
"call_id": "CA123",
|
||||
"trigger_uuid": "trigger-uuid-123",
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
def test_workflow_uuid_route_uses_scoped_lookup_and_shared_execution():
|
||||
|
|
@ -177,6 +194,7 @@ def test_workflow_uuid_route_uses_scoped_lookup_and_shared_execution():
|
|||
return_value=SimpleNamespace(id=55)
|
||||
)
|
||||
mock_db.create_workflow_run = AsyncMock(return_value=SimpleNamespace(id=601))
|
||||
mock_db.update_workflow_run = AsyncMock()
|
||||
|
||||
response = client.post(
|
||||
f"/public/agent/workflow/{workflow.workflow_uuid}",
|
||||
|
|
@ -207,6 +225,14 @@ def test_workflow_uuid_route_uses_scoped_lookup_and_shared_execution():
|
|||
assert create_kwargs["definition_id"] == 77
|
||||
assert "name" not in create_kwargs["initial_context"]
|
||||
assert not mock_db.get_draft_version.called
|
||||
mock_db.update_workflow_run.assert_awaited_once_with(
|
||||
run_id=601,
|
||||
gathered_context={
|
||||
"provider": "twilio",
|
||||
"triggered_by": "api",
|
||||
"call_id": "CA123",
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
def test_trigger_test_route_uses_draft_and_template_context_with_api_override():
|
||||
|
|
@ -262,6 +288,7 @@ def test_trigger_test_route_uses_draft_and_template_context_with_api_override():
|
|||
return_value=SimpleNamespace(id=55)
|
||||
)
|
||||
mock_db.create_workflow_run = AsyncMock(return_value=SimpleNamespace(id=501))
|
||||
mock_db.update_workflow_run = AsyncMock()
|
||||
|
||||
response = client.post(
|
||||
"/public/agent/test/trigger-uuid-123",
|
||||
|
|
@ -280,6 +307,15 @@ def test_trigger_test_route_uses_draft_and_template_context_with_api_override():
|
|||
assert create_kwargs["initial_context"]["age"] == 10
|
||||
assert create_kwargs["initial_context"]["rank"] == 2
|
||||
assert create_kwargs["initial_context"]["trigger_mode"] == "test"
|
||||
mock_db.update_workflow_run.assert_awaited_once_with(
|
||||
run_id=501,
|
||||
gathered_context={
|
||||
"provider": "twilio",
|
||||
"triggered_by": "api",
|
||||
"call_id": "CA123",
|
||||
"trigger_uuid": "trigger-uuid-123",
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
def test_workflow_uuid_test_route_uses_draft_and_template_context():
|
||||
|
|
@ -328,6 +364,7 @@ def test_workflow_uuid_test_route_uses_draft_and_template_context():
|
|||
return_value=SimpleNamespace(id=55)
|
||||
)
|
||||
mock_db.create_workflow_run = AsyncMock(return_value=SimpleNamespace(id=501))
|
||||
mock_db.update_workflow_run = AsyncMock()
|
||||
|
||||
response = client.post(
|
||||
f"/public/agent/test/workflow/{workflow.workflow_uuid}",
|
||||
|
|
@ -346,6 +383,75 @@ def test_workflow_uuid_test_route_uses_draft_and_template_context():
|
|||
assert create_kwargs["initial_context"]["age"] == 12
|
||||
assert create_kwargs["initial_context"]["rank"] == 2
|
||||
assert create_kwargs["initial_context"]["trigger_mode"] == "test"
|
||||
mock_db.update_workflow_run.assert_awaited_once_with(
|
||||
run_id=501,
|
||||
gathered_context={
|
||||
"provider": "twilio",
|
||||
"triggered_by": "api",
|
||||
"call_id": "CA123",
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
def test_trigger_route_still_returns_success_when_metadata_persistence_fails():
|
||||
app = _make_test_app()
|
||||
client = TestClient(app)
|
||||
|
||||
workflow = _active_workflow(trigger_path="trigger-uuid-123")
|
||||
provider = _provider()
|
||||
quota_mock = AsyncMock(
|
||||
return_value=SimpleNamespace(has_quota=True, error_message="")
|
||||
)
|
||||
|
||||
with (
|
||||
patch("api.routes.public_agent.db_client") as mock_db,
|
||||
patch("api.routes.public_agent.call_concurrency") as mock_concurrency,
|
||||
patch(
|
||||
"api.routes.public_agent.authorize_workflow_run_start",
|
||||
new=quota_mock,
|
||||
),
|
||||
patch(
|
||||
"api.routes.public_agent.get_default_telephony_provider",
|
||||
new=AsyncMock(return_value=provider),
|
||||
),
|
||||
patch(
|
||||
"api.routes.public_agent.get_backend_endpoints",
|
||||
new=AsyncMock(return_value=("https://api.example.com", "wss://ignored")),
|
||||
),
|
||||
):
|
||||
slot = object()
|
||||
mock_concurrency.acquire_org_slot = AsyncMock(return_value=slot)
|
||||
mock_concurrency.bind_workflow_run = AsyncMock()
|
||||
mock_concurrency.release_workflow_run_slot = AsyncMock()
|
||||
mock_concurrency.release_slot = AsyncMock()
|
||||
|
||||
mock_db.validate_api_key = AsyncMock(
|
||||
return_value=SimpleNamespace(id=7, organization_id=11, created_by=22)
|
||||
)
|
||||
mock_db.get_agent_trigger_by_path = AsyncMock(
|
||||
return_value=SimpleNamespace(
|
||||
workflow_id=workflow.id,
|
||||
organization_id=11,
|
||||
state="active",
|
||||
)
|
||||
)
|
||||
mock_db.get_workflow = AsyncMock(return_value=workflow)
|
||||
mock_db.get_default_telephony_configuration = AsyncMock(
|
||||
return_value=SimpleNamespace(id=55)
|
||||
)
|
||||
mock_db.create_workflow_run = AsyncMock(return_value=SimpleNamespace(id=501))
|
||||
mock_db.update_workflow_run = AsyncMock(side_effect=Exception("db down"))
|
||||
|
||||
response = client.post(
|
||||
"/public/agent/trigger-uuid-123",
|
||||
headers={"X-API-Key": "test-api-key"},
|
||||
json={"phone_number": "+15551234567"},
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
provider.initiate_call.assert_awaited_once()
|
||||
mock_db.update_workflow_run.assert_awaited_once()
|
||||
mock_concurrency.release_workflow_run_slot.assert_not_awaited()
|
||||
|
||||
|
||||
def test_trigger_route_rejects_when_concurrency_limit_reached():
|
||||
|
|
|
|||
|
|
@ -251,6 +251,10 @@ async def test_text_chat_session_creation_executes_initial_assistant_turn(
|
|||
assert workflow_run.initial_context == {
|
||||
"name": "explicit",
|
||||
"draft_only": "kept",
|
||||
"runtime_configuration": {
|
||||
"llm_provider": "openai",
|
||||
"llm_model": "gpt-4.1",
|
||||
},
|
||||
}
|
||||
assert "call_duration_seconds" in workflow_run.usage_info
|
||||
assert _log_texts(run_payload["logs"], "rtf-bot-text") == [
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue