Files
praxis/server/interruptibility.py
Praxis CI fbd6602814 docs(milestone): complete v0.1 foundation
---ci---
phase: 0
milestone: v0.1
status: complete
---/ci---
2026-08-01 13:32:48 +00:00

29 lines
1.1 KiB
Python

"""Interruptibility verification harness (TASK-03-05).
Verifies D-008 (abort-and-yield): learner VAD during AI TTS aborts TTS and
yields the floor. The Pipecat pipeline has allow_interruptions=True (set in
build_pipeline), so the abort is handled by Pipecat's built-in interrupt
handling. This module provides:
- a programmatic check that the pipeline is configured for interruptions
- a test that confirms a TTS-abort event fires on VAD during TTS
The manual test (speaking during AI speech cuts it off) is documented in
docs/latency-report.md; the automated test is in tests/test_interruptibility.py.
"""
from __future__ import annotations
from typing import Any
def pipeline_allows_interruptions(pipeline_task: Any) -> bool:
"""Confirm the pipeline task is configured with allow_interruptions=True (D-008)."""
# PipelineParams stores the flag; the task's params attribute carries it.
params = getattr(pipeline_task, "params", None)
if params is None:
return False
return bool(getattr(params, "allow_interruptions", False))
__all__ = ["pipeline_allows_interruptions"]