Async Job Queue Architecture for Podcasts Using Surreal-Commands in Open Notebook

Open Notebook implements a robust asynchronous job queue for podcast generation using Surreal-Commands, decoupling HTTP API requests from long-running audio processing through persistent job tracking and RecordID linking.

Open Notebook leverages Surreal-Commands to handle computationally expensive podcast generation tasks asynchronously. This architecture ensures that API endpoints remain responsive while the system processes AI-generated audio content in the background, with full visibility into job status and execution state.

Job Submission and Queue Registration

The async workflow begins when a client submits a generation request to the POST /podcasts/generate endpoint. This triggers the PodcastService.submit_generation_job method in api/podcast_service.py, which handles the initial orchestration.

The service performs the following steps:

1. Validates the requested episode and speaker profiles
2. Builds a command argument dictionary containing notebook content and configuration
3. Imports the command module (commands.podcast_commands) to ensure visibility within the Surreal-Commands framework
4. Invokes submit_command("open_notebook", "generate_podcast", args) to enqueue the job

The submit_command call returns a RecordID that uniquely identifies the job within Surreal-Commands. This identifier is immediately stored as a string (job_id) in the response payload, allowing clients to track progress without blocking on completion.

# Conceptual flow inside submit_generation_job (api/podcast_service.py)

args = {
    "episode_profile": request.episode_profile,
    "speaker_profile": request.speaker_profile,
    "content": notebook_content
}

# Import required to register command with Surreal-Commands runtime

import commands.podcast_commands  

job_id = submit_command("open_notebook", "generate_podcast", args)

# Returns: RecordID("job:open_notebook:generate_podcast:0001")

Command Execution and Persistence

When Surreal-Commands picks up the queued job, it executes the generate_podcast_command function defined in commands/podcast_commands.py. This function is decorated with @command("generate_podcast", app="open_notebook", …) to register it with the framework.

The command implementation follows a critical persistence pattern: it creates a PodcastEpisode record before initiating the heavy-weight podcast generation pipeline. This ensures the job remains trackable even if the process crashes during execution.

Specifically, the command stores the Surreal-Commands job identifier on the PodcastEpisode.command field:

# Inside generate_podcast_command (commands/podcast_commands.py)

from surreal_commands import ensure_record_id

episode = PodcastEpisode(
    name=input_data.args["episode_name"],
    command=ensure_record_id(input_data.execution_context.command_id),  # Links job to episode

    status="processing"
)
episode.save()  # Persisted before long-running work begins

# Heavy processing happens here...

# - Audio generation

# - Transcript creation

# - Outline generation

This early persistence strategy guarantees that every executing job has a corresponding database record, preventing orphaned processes and enabling reliable status tracking.

RecordID Linking and Data Model

The PodcastEpisode model in open_notebook/podcasts/models.py maintains the association between episodes and their background jobs through a specialized command field. This field stores the Surreal-Commands RecordID as a proper database reference rather than a plain string.

To ensure type consistency when writing to SurrealDB, the model overrides _prepare_save_data:

# In open_notebook/podcasts/models.py

class PodcastEpisode(BaseModel):
    command: Optional[RecordID] = None  # Points to Surreal-Commands job

    
    def _prepare_save_data(self):
        data = super()._prepare_save_data()
        if self.command:
            # Ensure proper RecordID format for SurrealDB

            data["command"] = ensure_record_id(self.command)
        return data

This bidirectional linking allows the system to traverse from an episode to its execution status, and from a job identifier back to its associated content.

Querying Job Status and Results

Clients poll for completion using the GET /podcasts/jobs/{job_id} endpoint, implemented in api/routers/podcasts.py. This router delegates to PodcastService.get_job_status, which calls await get_command_status(job_id) from the Surreal-Commands library.

The status query returns the complete execution context:

• Status: submitted, running, completed, or failed
• Result: Output data or error messages
• Timestamps: Creation, start, and completion times
• Metadata: Execution context and argument snapshots

When listing episodes, the system enriches the response with live job status via episode.get_job_detail(), which internally wraps get_command_status:

# Inside PodcastService.get_job_status (api/podcast_service.py)

async def get_job_status(self, job_id: str):
    status = await get_command_status(job_id)
    return {
        "status": status.status,
        "result": status.result,
        "error": status.error,
        "created_at": status.created_at,
        "completed_at": status.completed_at
    }

Handling Failures and Retries

The architecture includes a robust retry mechanism accessible via POST /podcasts/episodes/{episode_id}/retry. When a job fails or produces unsatisfactory results, this endpoint performs a clean slate operation:

1. Deletes the failed PodcastEpisode record
2. Removes associated audio files from storage
3. Re-submits a fresh job via PodcastService.submit_generation_job, generating a new Surreal-Commands RecordID

This approach guarantees that retried jobs receive new execution contexts, preventing contamination from previous failed states and ensuring idempotent retry operations.

# Retry endpoint implementation (api/routers/podcasts.py)

@router.post("/podcasts/episodes/{episode_id}/retry")
async def retry_episode(episode_id: str):
    # Delete old record and assets

    old_episode = await PodcastEpisode.get(episode_id)
    await delete_audio_file(old_episode.audio_file)
    await old_episode.delete()
    
    # Submit new job with same parameters

    new_job = await PodcastService.submit_generation_job(old_episode.config)
    return {"new_job_id": new_job["job_id"]}

Summary

The async job queue architecture for podcasts using Surreal-Commands provides:

• Decoupled processing: HTTP requests return immediately while heavy work continues in background processes
• Durable state tracking: RecordID linking between PodcastEpisode records and Surreal-Commands jobs ensures visibility into execution status
• Early persistence: Episode records are created before processing begins, preventing data loss during crashes
• Type-safe integration: The command field uses proper RecordID handling with ensure_record_id conversions
• Reliable retry semantics: Failed jobs can be retried with fresh execution contexts and new identifiers

Frequently Asked Questions

How does the system handle crashes during podcast generation?

The command implementation in commands/podcast_commands.py creates and saves the PodcastEpisode record with the command_id before starting the heavy-weight audio generation pipeline. This ensures the job record exists in the database even if the process crashes, allowing users to query the status and see a failed state rather than experiencing data loss.

What is the relationship between the job_id and the PodcastEpisode record?

The job_id returned by submit_command is stored in the PodcastEpisode.command field as a SurrealDB RecordID. This creates a bidirectional link: the episode knows which Surreal-Commands job is processing it, and the job context (via execution_context.command_id) can reference back to the episode record. The model overrides _prepare_save_data in open_notebook/podcasts/models.py to ensure this field is always properly formatted as a RecordID type.

Can multiple podcasts be generated simultaneously?

Yes. Each call to POST /podcasts/generate creates an independent Surreal-Commands job with a unique RecordID. The jobs execute concurrently based on the Surreal-Commands worker pool configuration, and each maintains its own state in the database. The PodcastEpisode records track individual job status, allowing simultaneous generation of multiple episodes without blocking.

How does the retry mechanism ensure clean execution?

The retry endpoint (POST /podcasts/episodes/{episode_id}/retry) performs a complete cleanup of the failed state. It deletes the old episode record, removes any partially generated audio files, and calls submit_generation_job to create a brand new Surreal-Commands job with a fresh RecordID. This prevents residual state from failed attempts from affecting the new execution.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}