Overview
VoicePact contracts progress through a well-defined lifecycle from initial creation through completion or cancellation. Each status transition is tracked with timestamps and audit logs, creating an immutable evidence trail for dispute resolution.
Contract Statuses
Defined in contract.py:25-32:
class ContractStatus(str, enum.Enum):
PENDING = "pending" # Draft, awaiting party confirmations
CONFIRMED = "confirmed" # All parties signed, not yet active
ACTIVE = "active" # Escrow locked, delivery in progress
COMPLETED = "completed" # Successful delivery, payment released
DISPUTED = "disputed" # Issue raised, requires mediation
CANCELLED = "cancelled" # Terminated before completion
EXPIRED = "expired" # Confirmation timeout exceeded
State Transition Diagram
┌─────────┐
│ PENDING │ (Created from voice or manual input)
└────┬────┘
│
│ All parties confirm
▼
┌───────────┐
│ CONFIRMED │ (Ready for escrow)
└────┬──────┘
│
│ Payment locked
▼
┌──────────┐
│ ACTIVE │ (Delivery phase)
└────┬─────┘
│
├──────────▶ [DISPUTED] ──▶ (Mediation/Resolution)
│
│ Delivery confirmed
▼
┌───────────┐
│ COMPLETED │ (Final state - payment released)
└───────────┘
PENDING ───timeout───▶ [EXPIRED]
PENDING/CONFIRMED ──▶ [CANCELLED] (by mutual agreement)
Lifecycle Phases
1. Creation (PENDING)
Trigger: Voice recording processed or manual contract created
Database model (contract.py:65-139):
class Contract(Base):
__tablename__ = "contracts"
id: Mapped[str] = mapped_column(String(50), primary_key=True)
transcript: Mapped[str] = mapped_column(Text)
contract_type: Mapped[ContractType] = mapped_column(
SQLEnum(ContractType),
default=ContractType.OTHER
)
terms: Mapped[dict] = mapped_column(JSON, default=dict)
contract_hash: Mapped[str] = mapped_column(String(128), unique=True)
total_amount: Mapped[Optional[Decimal]] = mapped_column(
Numeric(precision=15, scale=2)
)
status: Mapped[ContractStatus] = mapped_column(
SQLEnum(ContractStatus),
default=ContractStatus.PENDING
)
created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow)
expires_at: Mapped[Optional[datetime]] = mapped_column(DateTime)
Creation endpoint (contracts.py:61-131):
@router.post("/create", response_model=ContractResponse)
async def create_contract(
request: ContractCreateRequest,
background_tasks: BackgroundTasks,
contract_generator: ContractGenerator = Depends(get_contract_generator),
crypto_service: CryptoService = Depends(get_crypto_service),
db: AsyncSession = Depends(get_db)
):
# Generate contract hash for integrity
contract_hash = crypto_service.generate_contract_hash(
f"{request.transcript}:{str(sorted(request.terms.items()))}"
)
# Create contract
contract = Contract(
id=contract_id,
transcript=request.transcript,
contract_type=ContractType(request.contract_type),
terms=request.terms,
contract_hash=contract_hash,
status=ContractStatus.PENDING
)
db.add(contract)
# Create signature records for each party
for party_data in request.parties:
signature = ContractSignature(
contract_id=contract_id,
signer_phone=party_data.get('phone'),
signature_method="sms_confirmation",
status=SignatureStatus.PENDING
)
db.add(signature)
await db.commit()
# Send SMS confirmations in background
background_tasks.add_task(
send_contract_confirmations,
contract_id, request.parties, contract.terms, at_client
)
Contracts are created with a cryptographic hash of the transcript and terms. This hash is used to verify the contract hasn’t been tampered with.
2. Confirmation (PENDING → CONFIRMED)
Trigger: All parties sign via SMS, USSD, or OTP confirmation
Signature tracking (contract.py:173-216):
class ContractSignature(Base):
__tablename__ = "contract_signatures"
contract_id: Mapped[str] = mapped_column(
String(50),
ForeignKey("contracts.id", ondelete="CASCADE")
)
signer_phone: Mapped[str] = mapped_column(String(20), index=True)
signature_method: Mapped[str] = mapped_column(
String(20),
default="sms_confirmation"
)
signature_hash: Mapped[str] = mapped_column(String(128))
status: Mapped[SignatureStatus] = mapped_column(
SQLEnum(SignatureStatus),
default=SignatureStatus.PENDING
)
signed_at: Mapped[Optional[datetime]] = mapped_column(DateTime)
Confirmation endpoint (contracts.py:287-343):
@router.post("/{contract_id}/confirm")
async def confirm_contract(
contract_id: str,
phone_number: str,
db: AsyncSession = Depends(get_db)
):
# Find signature record
signature = await db.execute(
select(ContractSignature).where(
and_(
ContractSignature.contract_id == contract_id,
ContractSignature.signer_phone == phone_number
)
)
)
# Update signature
signature.status = SignatureStatus.SIGNED
signature.signed_at = datetime.utcnow()
# Check if all parties have signed
all_signatures = await db.execute(
select(ContractSignature).where(
ContractSignature.contract_id == contract_id
)
)
signed_count = sum(1 for sig in all_signatures if sig.status == SignatureStatus.SIGNED)
# If all signed, update contract to CONFIRMED
if signed_count == len(all_signatures):
contract.status = ContractStatus.CONFIRMED
contract.confirmed_at = datetime.utcnow()
await db.commit()
The contract transitions to CONFIRMED only when all parties have successfully signed. Partial signatures are tracked but don’t change the contract status.
3. Activation (CONFIRMED → ACTIVE)
Trigger: Escrow payment successfully locked
Once confirmed, the buyer initiates payment which is held in escrow:
# Contract status updated when payment is locked
if payment.status == PaymentStatus.LOCKED:
contract.status = ContractStatus.ACTIVE
await db.commit()
See Escrow Payments for payment flow details.
4. Completion (ACTIVE → COMPLETED)
Trigger: Delivery confirmed and escrow released
Status update (contracts.py:250-284):
@router.put("/{contract_id}")
async def update_contract(
contract_id: str,
request: ContractUpdateRequest,
db: AsyncSession = Depends(get_db)
):
contract = await db.execute(
select(Contract).where(Contract.id == contract_id)
)
if request.status == "completed":
contract.status = ContractStatus.COMPLETED
contract.completed_at = datetime.utcnow()
await db.commit()
Payment release is tied to status transition (see payments.py).
5. Dispute Resolution (ACTIVE → DISPUTED)
Trigger: Party raises issue via SMS/USSD
contract.status = ContractStatus.DISPUTED
# Escrow remains locked pending mediation
# Dispute details logged in audit trail
Disputed contracts freeze all state transitions until resolution. The audit log provides evidence for mediation.
6. Cancellation or Expiration
CANCELLED: Mutual agreement before activation
EXPIRED: No confirmations received within timeout window
# Set during contract creation (contract_generator.py:75)
expires_at = created_at + timedelta(seconds=settings.contract_confirmation_timeout)
# Background job checks for expiration
if contract.status == ContractStatus.PENDING and datetime.utcnow() > contract.expires_at:
contract.status = ContractStatus.EXPIRED
Contract Parties
Each contract involves multiple parties with defined roles (contract.py:142-170):
class PartyRole(str, enum.Enum):
BUYER = "buyer"
SELLER = "seller"
MEDIATOR = "mediator"
WITNESS = "witness"
class ContractParty(Base):
__tablename__ = "contract_parties"
contract_id: Mapped[str] = mapped_column(
String(50),
ForeignKey("contracts.id", ondelete="CASCADE")
)
phone_number: Mapped[str] = mapped_column(String(20), index=True)
role: Mapped[PartyRole] = mapped_column(SQLEnum(PartyRole))
name: Mapped[Optional[str]] = mapped_column(String(100))
organization: Mapped[Optional[str]] = mapped_column(String(100))
Audit Trail
Every state change is logged (contract.py:271-309):
class AuditLog(Base):
__tablename__ = "audit_logs"
contract_id: Mapped[str] = mapped_column(
String(50),
ForeignKey("contracts.id", ondelete="CASCADE")
)
action: Mapped[str] = mapped_column(String(50), index=True)
actor_phone: Mapped[Optional[str]] = mapped_column(String(20))
old_values: Mapped[Optional[dict]] = mapped_column(JSON, default=dict)
new_values: Mapped[Optional[dict]] = mapped_column(JSON, default=dict)
created_at: Mapped[datetime] = mapped_column(
DateTime,
default=datetime.utcnow
)
Audit logs record:
- Status transitions
- Payment events
- Signature confirmations
- Dispute filings
- Term modifications
Contract Status Endpoint
Check contract progress (contracts.py:346-394):
@router.get("/{contract_id}/status")
async def get_contract_status(contract_id: str, db: AsyncSession):
contract = await db.execute(
select(Contract).where(Contract.id == contract_id)
)
signatures = await db.execute(
select(ContractSignature).where(
ContractSignature.contract_id == contract_id
)
)
return {
"contract_id": contract_id,
"status": contract.status.value,
"created_at": contract.created_at.isoformat(),
"confirmed_at": contract.confirmed_at.isoformat() if contract.confirmed_at else None,
"signatures": [
{
"phone_number": sig.signer_phone,
"status": sig.status.value,
"signed_at": sig.signed_at.isoformat() if sig.signed_at else None
}
for sig in signatures
],
"progress": {
"signed": sum(1 for sig in signatures if sig.status == SignatureStatus.SIGNED),
"total": len(signatures),
"complete": contract.status in [ContractStatus.CONFIRMED, ContractStatus.ACTIVE, ContractStatus.COMPLETED]
}
}
Database Constraints
Data integrity is enforced at the database level (contract.py:133-139):
__table_args__ = (
Index("idx_contract_status_created", "status", "created_at"),
Index("idx_contract_type_status", "contract_type", "status"),
CheckConstraint("total_amount >= 0", name="check_positive_amount"),
CheckConstraint("created_at <= expires_at", name="check_valid_expiry"),
)
Best Practices
Lifecycle management tips:
- Expiration handling - Set reasonable confirmation timeouts (e.g., 24-48 hours)
- Audit everything - Log all state changes for dispute resolution
- Atomic transitions - Use database transactions to ensure consistency
- Signature verification - Validate SMS confirmations cryptographically
- Status checks - Poll
/status endpoint for real-time progress
- Immutability - Never modify
contract_hash or core terms after creation