debug-workflow-conversations

Debugging workflows from conversation logs

Every turn a fastWorkflow workflow executes is recorded — the user's message, the agent's plan, every intent-detection attempt with the classifier's confidence, every extracted parameter, every db_lookup correction, every validation verdict, every ask_user exchange, and the final answer. When a task fails, the log tells you which pipeline stage failed, and each stage has a specific fastWorkflow feature that fixes it. This skill is the map from log to fix.

1. Locate the database

One SQLite database per workflow, under the fastWorkflow state root:

from fastworkflow import state_paths
db_path = state_paths.observability_db("<workflow_folder>")
# typically ~/.local/state/fastworkflow/workflows/<workflow-name>/observability.sqlite3

If the file does not exist, the workflow has never run with observability on (the default under fastworkflow run, run_chatbot, and run_fastapi_mcp). Reproduce the failure first — fastworkflow run_chatbot <workflow> is the fastest way; every chat turn lands in this database immediately.

2. Read it — read-only, always

from fastworkflow.observability_store import ReadOnlyObservabilityStore
store = ReadOnlyObservabilityStore(db_path)

Never instantiate ObservabilityStore to inspect a database. That class is the writer: constructing it creates the file if missing, runs schema migration, and write-probes it. ReadOnlyObservabilityStore opens mode=ro connections and cannot mutate anything. Raw SQL is equally fine (the schema is documented in reference.md):

sqlite3 "file:$DB?mode=ro" "SELECT turn_key, status, success, user_message FROM turns ORDER BY turn_key DESC LIMIT 20"

3. Find the failing turn

store.list_turns(status="failed")                 # turn-level failures
store.list_turns(success=False)                   # any command in the turn failed
store.list_turns(command_name="cancel_order")     # turns that executed a command
store.list_turns(context="TodoList")              # substring match on entry context
turn  = store.get_turn(turn_key)                  # full row incl. record_json
spans = store.get_spans(turn_key)                 # the trace, ordered by start_ns