Tracing & Audit Logging

Tracing Integration

The trace_id field lets you correlate AxonFlow workflows with traces from external observability tools like Langsmith, Datadog, or OpenTelemetry.

Setting trace_id

Pass trace_id when creating a workflow:

curl -X POST http://localhost:8080/api/v1/workflows \
  -H "Content-Type: application/json" \
  -d '{
    "workflow_name": "code-review-pipeline",
    "source": "langgraph",
    "trace_id": "langsmith-run-abc123"
  }'

The trace_id is returned in the create response and preserved in all subsequent status queries.

SDK Usage

# Python
workflow = await client.create_workflow(
    CreateWorkflowRequest(
        workflow_name="research-agent",
        source=WorkflowSource.LANGGRAPH,
        trace_id="langsmith-run-abc123",  # External trace correlation
    )
)
# trace_id is available on the response
print(workflow.trace_id)  # "langsmith-run-abc123"

# LangGraph adapter
async with adapter:
    await adapter.start_workflow(trace_id="datadog-trace-xyz")

// Go
workflow, _ := client.CreateWorkflow(axonflow.CreateWorkflowRequest{
    WorkflowName: "research-agent",
    Source:       axonflow.WorkflowSourceLangGraph,
    TraceID:      "otel-trace-456",
})

Filtering by trace_id

List workflows by trace_id to find all executions associated with a specific external trace:

curl "http://localhost:8080/api/v1/workflows?trace_id=langsmith-run-abc123"

workflows = await client.list_workflows(
    ListWorkflowsOptions(trace_id="langsmith-run-abc123")
)

Audit Logging

Every workflow operation is automatically logged to the audit trail for compliance and debugging. All operations are recorded in the audit_logs table with the workflow ID as the request_id.

Operations Logged

Operation	Request Type	Description
Create Workflow	workflow_created	Workflow registration with name, source, and total steps
Step Gate Check	workflow_step_gate	Policy evaluation with decision, evaluated policies, and matched policies
Mark Step Completed	workflow_step_completed	Step completion with optional output and post-execution metrics
Complete Workflow	workflow_completed	Workflow finishes successfully
Abort Workflow	workflow_aborted	Workflow cancelled with reason

Querying Audit Logs

Use the SDK audit search methods to query workflow logs:

# Python SDK
from axonflow.types import AuditSearchRequest
from datetime import datetime, timedelta, timezone

# Get audit logs for a specific workflow
response = await client.search_audit_logs(
    AuditSearchRequest(
        start_time=datetime.now(timezone.utc) - timedelta(hours=1),
        limit=100,
    )
)

# Filter by workflow ID
for entry in response.entries:
    if entry.request_id == workflow.workflow_id:
        print(f"[{entry.timestamp}] {entry.request_type}")

// Go SDK
auditLogs, _ := client.SearchAuditLogs(ctx, &axonflow.AuditSearchRequest{
    StartTime: &startTime,
    Limit:     100,
})
for _, entry := range auditLogs.Entries {
    if entry.RequestID == workflowID {
        fmt.Printf("[%s] %s\n", entry.Timestamp, entry.RequestType)
    }
}

Verifying Audit Logs

See the workflow-policy examples for complete working examples that demonstrate audit log verification after workflow operations.

For more details on audit logging, see Audit Logging.