❌ Egress failed: TwirpError(code=resource_exhausted, message=egress minutes exceeded, status=429)

async def entrypoint(ctx: JobContext):
    <http://logger.info|logger.info>(f"--- Job Starting: {ctx.job.id} ---")
    await ctx.connect()

    # Get agent_id from job or room metadata
    agent_id = "default"
    user_id = None
    company_id = None
    try:
        job_metadata = json.loads(ctx.job.metadata or "{}")
        agent_id = job_metadata.get("agent_id", agent_id)
        user_id = job_metadata.get("user_id", user_id)
        company_id = job_metadata.get("company_id", company_id)
    except (json.JSONDecodeError, AttributeError):
        pass
    try:
        room_metadata = json.loads(ctx.room.metadata or "{}")
        agent_id = room_metadata.get("agent_id", agent_id)
        user_id = room_metadata.get("user_id", user_id)
        company_id = room_metadata.get("company_id", company_id)
    except (json.JSONDecodeError, AttributeError):
        pass

    if agent_id == "default":
        raise ValueError(
            "No agent_id found in metadata; pass via dispatch/agent-metadata"
        )

    <http://logger.info|logger.info>(f"🎯 Using agent_id: {agent_id}")

    # Load prompt & optional session timeout (in seconds) from DB
    instructions, timeout_seconds, model_config = await load_prompt_from_neon(agent_id)

    # Recording & object names
    base_name = make_session_base_name(
        room_name=ctx.room.name,
        job_id=ctx.job.id,
        agent_id=agent_id,
    )
    artifact_names = build_session_artifact_names(base_name)
    audio_object = artifact_names["audio_object"]
    transcript_object = artifact_names["transcript_object"]
    log_object = artifact_names["log_object"]
    transcript_path = artifact_names["transcript_path"]
    session_log_path = artifact_names["session_log_path"]

    # Capture per-session runtime logs to a dedicated file for archival.
    session_file_handler = add_session_file_handler(session_log_path)
    <http://logger.info|logger.info>(f"🧾 Session log file: {session_log_path}")

    with open(GOOGLE_APPLICATION_CREDENTIALS) as f:
        gcp_creds_json = json.dumps(json.load(f))  # Ensure string

    # 1. Start Egress directly to GCS (audio)
    egress_id = None
    try:
        egress_response = await ctx.api.egress.start_room_composite_egress(
            api.RoomCompositeEgressRequest(
                room_name=ctx.room.name,
                audio_only=True,
                file=api.EncodedFileOutput(
                    filepath=audio_object,  # used as object name
                    gcp=api.GCPUpload(  # <- GCS config
                        bucket=GCS_BUCKET_NAME,
                        credentials=gcp_creds_json,
                    ),
                ),
            )
        )
        egress_id = egress_response.egress_id
        <http://logger.info|logger.info>(f"✅ Egress to GCS: gs://{GCS_BUCKET_NAME}/{audio_object}")
    except Exception as e:
        logger.error(f"❌ Egress failed: {e}")

2026-04-30 12:33:23,934 INFO [universal-agent] 🧾 Session log file: /tmp/sellerity-roleplay-019dcdf4-79e8-7049-99f3-d82b78d2b34d-1777552398-02c0c0d7_AJ_n3QqvPhnMQjP_rp-fb6a9dd2a976_20260430T123323Z_session.log
2026-04-30 12:33:24,116 ERROR [universal-agent] ❌ Egress failed: TwirpError(code=resource_exhausted, message=egress minutes exceeded, status=429)
2026-04-30 12:33:24,277 INFO [universal-agent] 🚀 Starting AgentSession...