Guidance for AI assistants working on the Genesis project.
Genesis is an async Python library for FreeSWITCH Event Socket Layer (ESL) integration. Key components:
Inbound: ESL client (app connects to FreeSWITCH)Outbound: ESL server (FreeSWITCH connects to app)Consumer: Event consumption with decoratorsSession: Call session in outbound modeChannel: Channel abstractionProtocol: Base class for ESL clients
Poetry manages dependencies and virtual environments. Always use Poetry commands:
# Install dependencies
poetry install
# Add a dependency
poetry add package-name
# Add a dev dependency
poetry add --group dev package-name
# Run commands in the Poetry environment
poetry run <command>
# Activate shell
poetry shellImportant: Never use pip install directly. Always use poetry add or edit pyproject.toml and run poetry lock --no-update.
All code must be strictly typed and pass mypy validation:
# Check types
poetry run mypy
# Check specific file
poetry run mypy genesis/channel.pyRules:
- All functions, methods, and variables must have type hints
- Use
typingmodule types (List[str],Dict[str, Any],Optional[str]) - Use types from
genesis.typeswhen available (e.g.,HangupCause) - Fix all mypy errors before committing
Configuration: See [tool.mypy] in pyproject.toml:
- Checks
genesis/directory ignore_missing_imports = false(strict)check_untyped_defs = trueno_implicit_optional = true
Run tests with pytest:
# Run all tests
poetry run pytest
# Run specific test file
poetry run pytest tests/test_channel.py
# Run with verbose output
poetry run pytest -vvv
# Run specific test
poetry run pytest tests/test_channel.py::test_channel_stateTest Structure:
- Tests in
tests/directory - Fixtures in
tests/conftest.py - Mocks/stubs in
tests/doubles.py - Test data in
tests/payloads.py - Async tests use
pytest.mark.asyncioorasyncio_mode = "auto"
Configuration: See [tool.pytest.ini_options] in pyproject.toml:
asyncio_mode = "auto"(auto-detect async tests)timeout = 10(via pytest-timeout)addopts = "-vvv -x --full-trace --timeout=10"
CRITICAL: Never use asyncio.sleep() in tests:
- Tests must use events, awaitables, or proper synchronization primitives
- Use
wait_for_channels()helper or similar event-based waiting - Use
asyncio.create_task()and await the task completion - Use
asyncio.Event,asyncio.Condition, orasyncio.Futurefor coordination - Use
asyncio.wait_for()with events/conditions for timeout handling - NO EXCEPTIONS - all sleeps must be replaced with event-driven mechanisms
- Sleeps make tests flaky, slow, and unreliable
Validate across Python versions (3.10, 3.11, 3.12):
# Run tests on all Python versions
poetry run tox
# Run specific environment
poetry run tox -e py310
# List environments
poetry run tox listAlways run tox before PRs to ensure compatibility.
Use Black for formatting:
# Format code
poetry run black genesis/ tests/
# Check formatting (no changes)
poetry run black --check genesis/ tests/Version constraint: >=22.1,<25.0 (see pyproject.toml)
from typing import Optional, List, Dict, Any
from genesis.types import HangupCause
async def process_event(
event: ESLEvent,
channels: List[str],
metadata: Optional[Dict[str, Any]] = None
) -> Optional[HangupCause]:
...Rules:
- All imports must be at the top of the file
- Native Python imports first (standard library)
- Library imports second (third-party and project imports)
- Separate groups with a blank line
Example:
import asyncio
from typing import Optional, List, Dict, Any
from rich.console import Console
from rich.table import Table
from genesis.types import HangupCause
from genesis.channel import Channel
from genesis.consumer import ConsumerOrder:
- Standard library imports (sorted by length)
- Blank line
- Third-party library imports (sorted by length)
- Blank line (if project imports follow)
- Project imports (sorted by length)
- Always use
async/awaitfor I/O operations - Use
async withfor context managers - Handle
CancelledErrorappropriately - Use
asyncio.create_task()for concurrent operations
- Classes:
PascalCase(e.g.,ESLEvent,Session) - Functions/Methods:
snake_case(e.g.,send_command) - Constants:
UPPER_SNAKE_CASE(e.g.,TRACE_LEVEL_NUM) - Modules:
snake_case(e.g.,consumer.py)
The project uses a consolidated CI/CD pipeline in .github/workflows/main.yml:
Main Workflow (main.yml):
- Tests: Runs on all PRs and pushes to main (Python 3.10, 3.11, 3.12)
- Linter (Black)
- Type checker (Mypy)
- Unit tests (pytest)
- Update Contributors: Updates README.md contributors list (only on push to main)
- Bump Version: Automatically updates version to calendar format
YYYY.MM.DD(only on push to main) - Build & Deploy Docs: Builds and deploys documentation to GitHub Pages (only on push to main)
PyPI Workflow (pypi.yml):
- Publishes to PyPI when a release is published
Important Notes:
- The workflow automatically skips execution when commits are made by the bot (prevents infinite loops)
- Version bumping happens automatically using calendar format (year.month.day)
- All jobs run in sequence: tests → contributors → version → docs
The project uses calendar versioning (not semantic versioning):
- Format:
YYYY.MM.DD(e.g.,2026.01.24) - Automatically updated on each commit to main
- No manual version bumps needed
- Version is set in
pyproject.toml
Always prefer lists over long lines and tables:
- Use bullet lists for parameters, options, or multiple items
- Break long parameter descriptions into list items
- Use tables only when comparing multiple attributes side-by-side
- Keep lines readable (avoid horizontal scrolling)
Example - Good:
- `mode`: Ring mode (default: `PARALLEL`)
- `RingMode.PARALLEL`: Call all destinations simultaneously
- `RingMode.SEQUENTIAL`: Call destinations one at a timeExample - Bad:
- `mode`: Ring mode - `RingMode.PARALLEL` or `RingMode.SEQUENTIAL` (default: `PARALLEL`)All new features must include observability:
- Tracing: Use OpenTelemetry spans for all significant operations
- Create spans with
tracer.start_as_current_span()context manager - Add relevant attributes to spans (operation name, parameters, results)
- Record exceptions with
span.record_exception()
- Create spans with
- Metrics: Create counters and histograms for operations
- Use
meter.create_counter()for counting operations and results - Use
meter.create_histogram()for measuring durations - Include relevant attributes (mode, success/failure, error types)
- Use
- Pattern: Follow the same pattern used in
ChannelandProtocolmodules- Import
tracerandmeterfromopentelemetry - Create metrics at module level
- Wrap operations in spans with appropriate attributes
- Record metrics with operation results
- Import
Example:
from opentelemetry import trace, metrics
tracer = trace.get_tracer(__name__)
meter = metrics.get_meter(__name__)
operation_counter = meter.create_counter("genesis.feature.operations", unit="1")
operation_duration = meter.create_histogram("genesis.feature.duration", unit="s")
async def my_feature():
with tracer.start_as_current_span("feature.operation", attributes={...}) as span:
start_time = time.time()
try:
result = await do_work()
duration = time.time() - start_time
span.set_attribute("feature.result", "success")
operation_counter.add(1, attributes={"result": "success"})
operation_duration.record(duration)
return result
except Exception as e:
span.record_exception(e)
operation_counter.add(1, attributes={"result": "error"})
raiseAll OTel metric instruments live in genesis/protocol/metrics.py. Do not
re-declare an instrument that already exists there — duplicate instrument
creation for the same metric name trips static analysis and produces OTel SDK
warnings. Import the instrument (and the safe_add / safe_record helpers)
from genesis.protocol.metrics instead:
from genesis.protocol.metrics import (
calls_active_counter,
safe_add,
safe_record,
event_processing_duration,
)safe_add(counter, *args, **kwargs) and safe_record(histogram, *args, **kwargs)
swallow OTel/no-provider errors so a missing exporter never crashes the protocol.
genesis/protocol/lifecycle.py registers two event processors
(channel_lifecycle_processor, custom_subclass_processor) that emit
freeswitch.channel.* and freeswitch.sofia.* / freeswitch.callcenter.* /
freeswitch.conference.* / freeswitch.valet.* spans for the semantic
FreeSWITCH channel lifecycle. They run after the core protocol processors
(auth, command reply, disconnect) and only enrich telemetry — they never
consume events that route to user handlers. They are on by default; opt out
with GENESIS_TRACE_ESL_LIFECYCLE=0 / GENESIS_TRACE_CUSTOM_SUBCLASSES=0.
Emitted spans (non-exhaustive): freeswitch.channel.create, .progress,
.progress_media, .answer, .bridge, .unbridge, .hangup,
.hangup_complete, .destroy, .execute, .execute_complete, .codec,
freeswitch.call.update, freeswitch.sofia.transfer, freeswitch.sofia.register,
freeswitch.callcenter.info, freeswitch.conference.maintenance,
freeswitch.conference.cdr, freeswitch.valet.info.
Every channel lifecycle span carries sip.call_id (= the ESL
variable_sip_call_id header), the standard SIP Call-ID. This is a stable
per-call identifier that any other SIP observer of the same call will also
have, so it is the natural join key when correlating Genesis traces with
another system's traces of the same call. The join happens at the
observability backend (Grafana/Tempo), by filtering/grouping on sip.call_id
— not in code.
Cross-leg grouping: bridge spans carry bridge.a_uuid and bridge.b_uuid
(from Bridge-A-Unique-ID / Bridge-B-Unique-ID), so the a-leg and b-leg of a
call can be tied together at the backend.
The genesis.events.without_sip_call_id counter tracks channel events that
lack the correlation key (a correlation-gap signal). W3C traceparent /
X-Tracespan propagation is intentionally out of scope; the attribute join
is sufficient.
UUIDs go on spans only, never as metric attributes. Metric attributes use
low-cardinality enums/labels (channel.state, direction, hangup.cause,
application.name, bridge.result, transfer.type, loadbalancer.backend,
...). queue.depth is a span attribute (not a metric label) for the same
reason.
CRITICAL: Always run the full CI stack locally before opening a PR.
Before submitting a PR, you MUST run all CI checks locally:
- Formatting:
poetry run black --check genesis/ tests/ examples/- MUST pass - Types:
poetry run mypy genesis/- MUST pass with no errors - Tests:
poetry run pytest tests/ -v- MUST pass (all tests) - Multi-version:
poetry run tox- MUST pass on all Python versions (3.10, 3.11, 3.12)
Never open a PR without running these commands first. PRs that fail CI checks will be rejected.
Additional checks: 5. Dependencies: No unnecessary dependencies added 6. Documentation:
- README.md updated if needed
- All language versions updated (en, pt, ko) if documentation changed
- Follow documentation standards (prefer lists over long lines)
- Version: Version is automatically managed by CI/CD (calendar format)
Supported: Python 3.10, 3.11, 3.12
Constraint: >=3.10,<3.13 (see pyproject.toml)
- Documentation: https://otoru.github.io/Genesis/
- Repository: https://github.com/Otoru/Genesis
- PyPI: https://pypi.org/project/genesis/