Working with process_book

The process_book job orchestrates the book digitization pipeline using a work unit pattern. All stages are consolidated into this single job - there are no separate job packages.

Quick Reference

Key Locations

internal/jobs/process_book/
├── process_book.go       # Package entry, NewJob(), factory
└── job/
    ├── job.go            # Job struct, Start(), OnComplete()
    ├── types.go          # WorkUnitInfo, constants
    ├── state.go          # MaybeStartBookOperations(), triggers
    ├── finalize.go       # Finalize-ToC inline (discover, gap analysis)
    ├── structure.go      # Common-structure inline (extract, polish)
    ├── link_toc.go       # ToC linking agents
    └── [stage].go        # One file per stage

internal/jobs/common/
├── state.go              # BookState, PageState, OperationState
├── load.go               # LoadBook()
├── persist.go            # Persistence helpers (async-first)
├── reset.go              # Reset helpers for crash recovery
└── [helpers].go          # Shared utilities

Core Patterns

Work Unit Lifecycle

// 1. Create and register
unit := j.CreateXWorkUnit(ctx, pageNum, state)
j.RegisterWorkUnit(unit.ID, WorkUnitInfo{...})

// 2. Worker executes (external)

// 3. Handle completion
func (j *Job) OnComplete(ctx, result) {
    info, _ := j.GetWorkUnit(result.WorkUnitID)
    switch info.UnitType {
    case WorkUnitTypeX:
        return j.HandleXComplete(ctx, info, result)
    }
}

// 4. Clean up
j.RemoveWorkUnit(result.WorkUnitID)

State Persistence (Async-First)

// Async writes (default - non-blocking, fire-and-forget)
state.SetComplete(true)                // In-memory first
sink.Send(defra.WriteOp{...})          // Async to DB

// Sync only for critical creates (need DocID back)
result, _ := sink.SendSync(ctx, op)    // Blocking
j.TocDocID = result.DocID

Important: Agent state persistence is fully async. No intermediate saves during agent loops - crash recovery restarts agents from scratch.

Cascading Triggers (state.go)

func (j *Job) MaybeStartBookOperations(ctx) []jobs.WorkUnit {
    if j.SomeThreshold() && j.Book.SomeOp.CanStart() {
        j.Book.SomeOp.Start()
        return []jobs.WorkUnit{j.CreateSomeWorkUnit(ctx)}
    }
    // ... more triggers
}

OperationState API

op.CanStart()   // true if not started
op.Start()      // NotStarted → InProgress
op.Complete()   // InProgress → Complete
op.Fail(max)    // Increment retry, maybe → Failed
op.IsDone()     // Complete or permanently Failed

Stage Categories

See [stages.md](references/stages.md) for implementation patterns.

Debugging

Via API

shelf api jobs status <book-id>      # Stage progress
shelf api books get <book-id>        # Book state
shelf api agent-logs list --job-id X # Agent logs (if debug enabled)

Via DefraDB

{ Page(filter: {book_id: {_eq: "<id>"}}) { page_num ocr_complete } }
{ Book(filter: {_docID: {_eq: "<id>"}}) { metadata_complete toc_finalize_complete } }
{ AgentState(filter: {book_id: {_eq: "<id>"}}) { agent_id agent_type complete } }

See [debugging.md](references/debugging.md) for full query reference.

Common Tasks

Add a new stage

1. Define WorkUnitTypeX constant in types.go
2. Add state fields to PageState or BookState
3. Create handler file with CreateX and HandleXComplete
4. Add case to OnComplete() switch
5. Add trigger in handler or MaybeStartBookOperations()
6. Add persistence helper in common/persist.go (use async SendToSink)

See [adding-stages.md](references/adding-stages.md) for full guide.

Debug a stuck stage

1. Query DB for precondition states
2. Check started / complete flags
3. Enable debug: Config.DebugAgents = true
4. Check agent logs if agent-based

Agent state and crash recovery

• Agent states are persisted once at creation (async)
• No intermediate saves during the agent loop (performance optimization)
• On crash, agents restart from scratch (acceptable tradeoff)
• Use DeleteAgentStateByAgentID for cleanup (not DocID-based)