Skip to main content

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:
  1. Expiration handling - Set reasonable confirmation timeouts (e.g., 24-48 hours)
  2. Audit everything - Log all state changes for dispute resolution
  3. Atomic transitions - Use database transactions to ensure consistency
  4. Signature verification - Validate SMS confirmations cryptographically
  5. Status checks - Poll /status endpoint for real-time progress
  6. Immutability - Never modify contract_hash or core terms after creation

Build docs developers (and LLMs) love