How the Open Notebook Source Model Tracks Processing Status Using the Command Field

The Open Notebook Source model tracks asynchronous processing status by storing a SurrealDB job ID in its command field, then exposing helper methods that query the surreal-commands system to return real-time status and progress metadata.

In the lfnovo/open-notebook repository, every document, PDF, or website ingested into the system is represented by the Source Pydantic model. This model uses a dedicated command field to maintain a persistent link to background processing jobs, enabling real-time status tracking without exposing internal job identifiers to API consumers.

The Command Field Architecture

The command field is defined in open_notebook/domain/notebook.py (lines 54-94) as an optional union type that accepts either a string or SurrealDB RecordID:

command: Optional[Union[str, RecordID]] = Field(
    default=None,
    description="Link to surreal-commands processing job"
)

When a source enters the background workflow, the process_source command—defined in commands/source_commands.py (lines 86-99)—generates a unique job ID and persists it to this field:

# commands/source_commands.py – after loading the Source

source.command = ensure_record_id(input_data.execution_context.command_id)
await source.save()

This design decouples the domain model from command execution details while maintaining a queryable reference to the underlying job state.

Tracking Status with Built-in Methods

The Source model provides two asynchronous convenience methods that abstract interaction with the surreal_commands.get_command_status helper.

get_status() Method

The get_status() method returns the high-level job state as a string—such as queued, running, completed, or failed—or None if no command is attached. It calls get_command_status(str(self.command)) and extracts the status field.

get_processing_progress() Method

For richer telemetry, get_processing_progress() returns a dictionary containing status timestamps, error messages, and raw execution results. It retrieves the full command record, then extracts execution_metadata to build a detailed progress payload.

Implementation in the Processing Workflow

The link between a source and its processing job is established during the LangGraph workflow execution invoked by process_source. The workflow orchestrated in open_notebook/graphs/source.py creates the command context, which the command handler then attaches to the source record using ensure_record_id() to guarantee proper RecordID formatting.

Practical Usage Examples

Checking Status in API Endpoints

Service layers can leverage the model's methods to expose status via REST endpoints. In api/sources_service.py, a typical implementation looks like:

# api/sources_service.py

async def get_source_status(source_id: str) -> str:
    src = await Source.get(source_id)
    if not src:
        raise HTTPException(status_code=404, detail="Source not found")
    status = await src.get_status()
    return status or "no-job"

Displaying Progress in Frontend Applications

Frontend clients can poll for detailed progress metrics. A React hook using Zustand might fetch the extended progress payload:

// frontend/src/state/source.ts
export const useSourceProgress = (sourceId: string) => {
  const query = useQuery(['sourceProgress', sourceId], async () => {
    const resp = await fetch(`/api/sources/${sourceId}/progress`);
    return resp.json(); // {status, started_at, completed_at, error, result}
  });
  return query;
};

Manually Attaching Commands for Administration

For rare administrative tasks or debugging, you can manually link a source to a command ID in a REPL:

# In a REPL or admin script

src = await Source.get("source:12345")
src.command = ensure_record_id("command:embed_source:abcdef")
await src.save()
print(await src.get_status())   # → "queued"

Summary

• The Source model in open_notebook/domain/notebook.py tracks processing through an optional command field that stores SurrealDB RecordIDs.
• The surreal-commands system writes job IDs to this field during the process_source workflow defined in commands/source_commands.py.
• Two async helper methods—get_status() and get_processing_progress()—abstract command queries and return standardized status or detailed metadata.
• API layers in api/sources_service.py consume these methods to expose processing state without leaking internal job identifiers.
• Frontend applications can poll endpoints to retrieve real-time status updates for progress bars and error alerts.

Frequently Asked Questions

What happens if the command field is null?

If no background job has been attached, both get_status() and get_processing_progress() return None. This allows the application layer to display a "no-job" or "pending" state as appropriate without handling exceptions.

Can I use string IDs instead of RecordID objects for the command field?

Yes. The field accepts Optional[Union[str, RecordID]], though the codebase consistently uses ensure_record_id() to normalize inputs to proper SurrealDB RecordID instances before storage to ensure database compatibility.

How does the Source model handle failed commands?

The get_processing_progress() method surfaces error messages stored in the command's execution metadata. When get_status() returns "failed", the detailed progress payload contains the specific error message and stack trace from the background workflow.

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