Journey Engine

The Journey engine is Purple8's built-in AI workflow orchestration layer. It is not a database feature — it is a complete framework for stateful, auditable, AI-driven workflows backed by the graph.

The key insight

Journey is not a plugin for LangChain or CrewAI. It is the alternative. Every stage transition is a graph edge. Every AI decision is an immutable audit entry. Every workflow has SLA enforcement built in.

"Journey IS your agent framework — with graph memory, hybrid search, audit trails, human-in-the-loop oversight, and SLA enforcement built in."

Core concepts

Concept	Description
JourneyDefinition	A template for a repeatable process — a list of stages, each with an owner, optional SLA, and optional AI step config
JourneyInstance	A running instance of a definition tied to a specific entity (a customer, a document, a case)
StageSpec	One stage in a journey — with SLAPolicy and optional AIStepConfig
SLAPolicy	Warn after N seconds, breach after M seconds — enforced by a background thread
AIStepConfig	When to invoke the LLM, which actions it can take, whether to auto-execute
JourneyAIAdvisor	The LLM decision-maker — fires on every configured trigger with full audit context

Define a journey

from purple8_graph import GraphEngine
from purple8_graph.journey import (
    JourneyEngine, JourneyAIAdvisor,
    StageSpec, SLAPolicy, AIStepConfig, AIStepTrigger, AIActionType,
)
from purple8_graph.genai import OpenAIProvider

engine = GraphEngine("./data")
je = JourneyEngine(engine)

je.define_journey(
    journey_type="document_review",
    stages=[
        StageSpec("submitted"),
        StageSpec(
            "classified",
            owner_system="ClassifierService",
            sla=SLAPolicy(warn_after_seconds=1800, breach_after_seconds=3600),
            ai_step=AIStepConfig(
                trigger=AIStepTrigger.ON_ENTER,   # fires immediately on stage entry
                auto_execute=True,                 # LLM decision runs automatically
                allowed_actions=[
                    AIActionType.SUGGEST_ADVANCE,
                    AIActionType.FLAG,
                    AIActionType.ANNOTATE,
                ],
                extra_context={"policy": "Flag documents containing PII for legal review"},
            ),
        ),
        StageSpec(
            "reviewed",
            owner_system="ReviewTeam",
            sla=SLAPolicy(warn_after_seconds=7200, breach_after_seconds=14400),
            ai_step=AIStepConfig(
                trigger=AIStepTrigger.ON_SLA_WARN,  # fires when SLA is at risk
                auto_execute=False,                  # human must confirm
            ),
        ),
        StageSpec("approved"),
        StageSpec("rejected"),
    ],
    description="End-to-end document review with classification and legal flag",
)

Register the AI advisor

provider = OpenAIProvider(api_key="...")
advisor = JourneyAIAdvisor(je, provider)
je.register_ai_advisor(advisor)

Start and advance an instance

Normal advances visit non-terminal stages in definition order. They cannot skip or revisit a review stage. Fast-tracking requires an explicit force=True (REST: "force": true), authenticated journey:admin permission, and produces an [ADMIN OVERRIDE] audit note. Trusted embedded SDK callers can also request that explicit override.

REST and MCP audit identities come from the authenticated user, not a submitted actor string. SDK callers without an authenticated request can still supply system actor names. HITL claims are bound to that identity; another user cannot complete, reject or escalate a claimed task. Completing a review releases only the current gate and does not authorize skipping subsequent stages.

Required reviews fail closed if their queue or task lookup is unavailable. Entering a required review (including the first stage) creates its HITL task. Journey records are not editable through generic graph CRUD or Cypher.

# Start a journey for a specific document
instance = je.start(
    journey_type="document_review",
    entity_id="doc_abc123",
    metadata={"filename": "contract_2026_q1.pdf", "submitter": "legal@acme.com"},
)

# Advance to the next stage (called by whatever system completes this step)
je.advance(
    instance_id=instance.instance_id,
    to_stage="classified",
    actor="ClassifierService",
    notes="Classified as: Contract. PII detected: Yes.",
)
# → AIStepConfig.trigger=ON_ENTER fires here automatically
# → JourneyAIAdvisor receives full context, decides to FLAG
# → AI_ADVISED edge written to graph with reasoning

What JourneyAIAdvisor receives

On every advise() call, the advisor receives:

• Full audit history — every prior stage transition, every prior AI decision, every annotation
• Current stage — where the instance is now, who owns it
• SLA health — on-time, at-risk, or in breach
• Business metadata — anything passed at start() or advance()
• Workflow definition — the complete stage graph with names, SLAs, owners

The LLM returns one of:

Action	Effect
suggest_advance	Recommends moving to a named stage (human confirms unless auto_execute=True)
force_advance	Advances immediately (only when auto_execute=True)
cancel	Cancels the instance
flag	Annotates with an alert and holds for human review
hold	Pauses the instance (status → STALLED) without a breach
annotate	Writes a note to the audit trail — no state change
no_action	LLM determined no action is warranted

Every action, including no_action, is written to the audit trail as an AI_ADVISED edge with the LLM's reasoning.

Read the audit trail

# Full immutable audit history — graph edges, append-only
for event in je.audit(instance.instance_id):
    print(f"[{event.timestamp}] {event.from_stage} → {event.to_stage}")
    print(f"  actor: {event.actor}  notes: {event.notes}")

# Current status
status = je.status(instance.instance_id)
print(status.current_stage, status.sla_status)  # e.g. "classified", "AT_RISK"

SLA enforcement

The SLAMonitor background thread wakes every poll_interval seconds and checks active instances in its tenant-scoped engine. When an SLA fires:

• A warning writes an append-only SLA_WARNED edge.
• A breach writes an append-only SLA_BREACHED edge and updates the instance status to STALLED.
• A CDC ChangeEvent is published on the EventBus, if configured.
• A registered advisor runs in a background thread when the stage's trigger matches ON_SLA_WARN or ON_SLA_BREACH, respectively. Register it with je.register_ai_advisor(advisor) before monitoring.

SLA event IDs are deterministic for the tenant, instance, stage entry and event type. Retrying an interrupted audit write does not create another SLA edge. Monitor recreation reseeds recorded events and repairs an interrupted breach status update. Re-entering a stage starts a new SLA event identity.

Startup reseeding and polling stream instance models through the native tenant-scoped label iterator rather than collecting a full instance list. Scans check cancellation between records, including records excluded by the active-status filter, and close their iterators on cancellation or failure. The public list_instances() API still returns a list. Historical warning/breach reseeding also streams typed tenant-scoped adjacency in native pages of at most 128 records, rather than materializing each instance's complete edge history. It closes the active history iterator on cancellation or failure. This bounds retained edge models, not the separate historical fired-event key cache.

Embedded callers can pass a nonblocking cancellation predicate: je.check_slas(cancelled=stop_event.is_set). On-demand checks initialize their monitor before scanning history, so runtime close is not blocked by holding the Journey lock across that scan. An interrupted initial seed must finish on a later check before new effects run. Unreadable history blocks effects and surfaces an error until a subsequent seed succeeds; it must not cause old warnings or advisories to replay.

Automatic advisories retain the caller's actor context. A delayed response cannot mutate an instance whose stage entry or status has changed, and AI advancement does not bypass required human review. AI_ADVISED edges record ai_executed and ai_error so failed automatic actions remain inspectable. Audit-storage failures propagate instead of returning a successful advisory. For automatic work they are counted as failed dispatches and exposed in runtime status and the signed console. An action or audit write may already have committed when its acknowledgement fails; inspect the instance and audit before retrying. There is no automatic rollback or safe whole-advisory retry guarantee.

Durable Journey AI Outbox (v0.72.0) & Delivery Boundary

In v0.72.0, stages and SLA events (ADVANCED_TO, SLA_WARNED, SLA_BREACHED) that trigger an AI step write ai_dispatched=False onto the durable audit edge before invoking the advisor, and update it to ai_dispatched=True (with ai_dispatched_at) once the advisor completes. If the process crashes (SIGKILL) after persisting the transition or SLA edge but before the advisor completes, calling je.replay_pending_ai_dispatches() on recovery automatically replays any pending ai_dispatched=False steps.

Note that LLM provider execution itself remains at-least-once across hard crashes (an interrupted external LLM call is re-invoked on recovery if completion was not recorded). Missing-advisor or disabled-provider skips do not count as started dispatches, and external provider latency is bounded by provider timeouts rather than storage locks.

SLA operational controls

Operation	REST	MCP journey.sla_alerts	Permission
Monitor health	GET /v1/journeys/sla/health	operation="health"	journey:read
Warning/breach history	GET /v1/journeys/{instance_id}/sla/history?limit=50&offset=0	operation="history", instance_id, optional limit/offset	journey:read
Acknowledge one event	POST /v1/journeys/{instance_id}/sla/acknowledge	operation="acknowledge", instance_id, event_id, optional notes	journey:write
Reset a stalled clock	POST /v1/journeys/{instance_id}/sla/reset	operation="reset", instance_id, reason, expected_stage_entered_at	journey:admin

Omitting operation preserves the existing breach-list MCP behavior. SDK equivalents are sla_monitor_health(), sla_history(), acknowledge_sla() and reset_sla().

Health reports local check state, last outcome/timestamps, visits, errors, initial-seed completion and whether the local background thread is alive. An uninitialized monitor is not presented as a successful check. Server-managed runtimes additionally report actual scheduler state and interval, last sweep timestamps and a sanitized failure type. A failed scheduler tick is logged, reported as degraded and retried on a later tick. Scheduler liveness is not proof that every tenant check succeeded; consult the local check outcome. Standalone/unmanaged runtimes report scheduler health as not_reported. Provider connectivity is not checked by this health operation.

History requires an instance ID and returns at most 200 events per page (limit defaults to 50). It streams typed native adjacency rather than materializing the entire audit. Order is warning events followed by breach events, each in native adjacency order, not newest-first or a snapshot. Concurrent changes may affect offset pages. Each event includes its immutable ID and any acknowledgement. Acknowledging an event appends one SLA_ACKNOWLEDGED audit edge; retries return the existing acknowledgement. It does not resume a stalled instance, hide the original event or reset a clock.

Reset is deliberately narrower than an arbitrary edit: only a STALLED instance may be resumed. Supply the current stage_entered_at from journey.status/the status endpoint as expected_stage_entered_at and a non-empty reason. The reset sets the same stage back to ACTIVE with a fresh clock; it does not advance a stage or resolve/skip a HITL task. Stale occurrence requests are rejected. Retrying the same actor/reason/occurrence returns already_applied=true without moving the clock again.

The new clock and its audit receipt are stored in the same instance mutation. The immutable SLA_RESET audit edge is an idempotent projection of that receipt. If projecting it fails, the request fails explicitly; a retry or subsequent SLA check reconstructs the missing edge before further SLA effects. Preserve reserved Journey node properties when backing up or restoring workflows. This is a single-record recovery protocol, not a crash-atomic multi-record transaction or an exactly-once AI-delivery claim. Malformed receipts or conflicting projections fail closed and produce degraded check health. New receipts use version 1; the initial unversioned format remains readable. Unknown or malformed versions are rejected. Receipt fields, including legacy SDK actor/notes and stored stage/clock values, are validated before a new clock is persisted, so invalid inputs cannot leave a partially applied reset.

Clock-reset authorization change

The legacy POST /v1/journeys/{instance_id}/resume and authenticated SDK resume() now also require journey:admin, because they reset the same SLA clock. journey:write alone is no longer sufficient. Ordinary trusted SDK calls without an authenticated actor retain their existing embedded-use contract. Legacy resume receives the same persisted receipt/audit-repair protection.

In LCNC, open Journeys → SLA Alerts to inspect monitor/scheduler health, then open a journey to page through SLA events. Writers can acknowledge an event; admins see a reason-required reset form for stalled instances. Controls are permission-gated and all actions use the signed session's tenant and actor.

For shared storage, pass the tenant-scoped engine returned by your tenant manager and its matching tenant_id to JourneyEngine. Journey definition reloads, instance filters, audit adjacency and HITL task reads use that logical engine; do not substitute the raw shared storage object.

REST, MCP and the human console

The server resolves Journey and HITL operations from the authenticated tenant, not a tenant supplied in the request body or query string. This applies to /v1/journeys, the journey.* MCP tools that call those endpoints, and the LCNC Journey pages. A missing tenant is rejected; an unavailable or incorrectly bound runtime never falls back to the default tenant. Recovery-required tenants return HTTP 503 rather than using a cached runtime.

Requests and lifespan cleanup are bound to the application that owns the runtime, not a replacement process-global default. Closing a second application cannot redirect the surviving application's Journey requests to its closed manager. This protects runtime identity, not complete multi-server isolation: process-wide CDC, scheduler and executor services remain shared. Deploy independent production servers separately and never use multiple writers for one persistent data path.

Operation	Required permission
List definitions, instances, status, audit and HITL tasks	JOURNEY_READ
Start, advance, cancel, resume, record outcomes or resolve HITL tasks	JOURNEY_WRITE
Define or overwrite journey types	JOURNEY_ADMIN
Read automatic AI configuration/status	JOURNEY_READ
Enable or disable automatic AI	JOURNEY_ADMIN

LCNC additionally requires ADMIN_SETTINGS, a valid signed session and a live user whose tenant still matches the session. Console access alone does not grant workflow mutation privileges. Audit actors come from authenticated identity; operator-supplied actor names cannot impersonate someone else.

The console's stage form persists native SLAPolicy and required HITL settings. Checking only a later stage's HITL box gates that stage, including after removing earlier form rows. Resolving its task is still necessary before advancing.

Server-managed automatic AI

The HTTP server registers a tenant-scoped advisor when it creates a Journey runtime, including unattended SLA runtimes admitted after restart. Automatic execution is disabled by default, including for existing stage definitions. An administrator must explicitly opt in for each tenant; defining ai_step alone does not enable provider calls.

1. Save a generation-capable provider/model in the existing tenant LLM settings. SentenceTransformer-only configurations cannot generate advice.

2. Read GET /v1/journeys/runtime/ai.

3. Enable with PUT /v1/journeys/runtime/ai and JSON {"enabled": true}. The setting persists in that tenant's graph. Use false to disable.

4. Configure a stage's native ai_step, for example:

   {
     "name": "review",
     "ai_step": {
       "trigger": "on_enter",
       "auto_execute": false,
       "allowed_actions": ["suggest_advance", "no_action"],
       "extra_context": {}
     }
   }

Supported triggers are on_enter, on_sla_warn and on_sla_breach. system_prompt is optional. The REST stage schema now persists these native fields. Obsolete per-stage fields such as provider, model, prompt, output_field and on_error are rejected rather than silently ignored. Providers and credentials belong in tenant LLM settings, not stage definitions.

Agents use journey.list(runtime_status=true) for status and journey.define(operation="configure_ai", ai_enabled=true) for enablement. Ordinary journey.define(journey_type=..., stages=...) is unchanged. The configuration operation retains the tool's admin/super-admin role restriction; the REST permission remains JOURNEY_ADMIN. The LCNC Journey page displays runtime status and exposes an enable/disable control only to users with that permission, in addition to its normal signed-session and ADMIN_SETTINGS checks.

Status includes tenant identity, enablement, advisor registration, provider/model names, a sanitized last provider error code, and the best-effort delivery boundary. automatic_dispatch reports started, completed, failed, inflight, last_started_at, last_finished_at and a sanitized last_error_type. These are observations of automatic worker attempts in this runtime process, not durable delivery totals; they reset when the runtime is replaced. Direct manual advise() calls are not included. A persisted advisory with a nonempty error counts as failed (AdvisoryResultError), not completed; failed audit acknowledgements and thread-start failures also count as failed. The console shows these counters, timestamps and the last failure type without raw error messages. A later successful attempt clears the last failure type but preserves the cumulative failed count within that runtime. ready means settings are present, not that connectivity or credentials have been tested; provider_health is explicitly not_checked. Missing settings and embedding-only local providers report unavailable states. Settings are reloaded for each call, so provider changes do not require recreating the runtime. Unavailable encrypted keys fail closed instead of selecting ambient credentials.

Enablement applies to future triggers; missed advisories are not backfilled. Disabling does not cancel a provider request already sent, but its delayed response cannot automatically mutate the workflow after disablement. Closing the runtime also rejects late effects and audit writes. A sweep-only runtime waits for its dispatched advisory threads before releasing native storage; this does not establish a finite provider/shutdown timeout.

Before applying an automatic effect, a server-managed advisor reloads persisted authorization rather than trusting the copied request permissions. It checks the user's current status, tenant and roles, and intersects those permissions with the original request's grants: a later privilege increase cannot elevate an already pending action. When present, the original JWT session and expiry are checked again; signed-console expiry is also carried into the delayed action.

API-key calls retain only the key's identifier and a digest of its authorization constraints, not its credential. Execution checks the live owner, activation, expiry, explicit permission cap and unchanged tool/label/application/classification restrictions. Changed restrictions invalidate pending effects even if the new policy is broader; submit a new request under the new policy instead. An explicit empty permission set remains empty across auth-store reloads, distinct from null (inherit the owner's permissions).

Missing, corrupt or unavailable authorization metadata fails closed. Rejections appear in the AI_ADVISED audit's ai_error, with ai_executed=false; they are not reported as successful mutations. Runtime status advertises execution_authorization="persisted_actor_or_internal_sla". Unattended internal SLA checks have no requesting user and use the tenant's explicit opt-in and stage policy, not a remembered human's grants. Embedded SDK advisors retain their existing trusted-caller behavior. This is a pre-effect authorization check, not an atomic transaction spanning auth-file changes and graph effects, and it does not provide a durable delivery queue.

Embedded SDK applications still register their own advisor explicitly; the server does not reconstruct arbitrary SDK provider objects. Server controls return an explicit conflict if mounted around an unmanaged SDK runtime.

For an embedded application, prefer tenant_manager.get_journey_engine(tenant_id) to reuse the tenant's runtime and HITL queue. The manager closes its runtime on suspension/deletion and replaces it when the underlying engine changes.

The HTTP server runs one SLA scheduler for persisted active tenants, including tenants that have not received a request since restart. P8G_JOURNEY_SLA_INTERVAL sets its interval in seconds (default 30); 0 or a negative value disables automatic checks. Read-only servers and read-only tenant snapshots do not run mutating SLA checks. Embedded callers can explicitly invoke tenant_manager.check_journey_slas().

Sweep-only stores and Journey/HITL runtimes are opened one tenant at a time and released afterward. Existing application-owned stores are preserved. A concurrent application admission promotes a temporary store/runtime so the sweep does not close it. Empty tenants do not retain runtimes, and missing dedicated storage is not provisioned merely for an SLA check. Newly discovered runtimes receive the server's CDC event bus. Embedded callers can supply event_bus and stop_event keyword arguments to check_journey_slas.

Scheduler/caller cancellation is checked between tenants and inside an ongoing tenant's instance scan and history reseeding. This is cooperative cancellation, not preemption of an in-flight storage operation or provider call. An uncertain native store close is logged and blocks readmission until restart/recovery; an uncertain physical shared-store close blocks all shared-store admissions.

SLA working sets and connector scope

Instance-model and historical edge-model retention during startup and polling are bounded by streaming; neither retains complete scan-result lists. The one-temporary-tenant admission bound still does not bound the historical fired-event key cache, the tenant catalog snapshot, or shutdown latency inside a blocking operation. Polling still traverses the instance label and filters status; this is not a paged active-only property-index scan or a qualified throughput claim for large completed-workflow histories. Those resource bounds and durable AI dispatch remain production qualification gates; operational health/history/acknowledgement/reset are exposed as above. A restart sweep does not restore an arbitrary SDK advisor registered only in process memory. Server-managed bindings are reconstructed from persisted tenant opt-in and the current tenant LLM settings. Per-event advisory threads remain unbounded; durable delivery, bounded admission and shutdown are not implied by provisioning.

The legacy signed webhook receiver is bound to its configured tenant; it is not a multi-tenant connector registry. Other tenants cannot read its connector configuration. Do not interpret Journey tenant routing as per-tenant webhook provisioning.

# SLA configuration
StageSpec(
    "reviewed",
    sla=SLAPolicy(
        warn_after_seconds=7200,    # 2 hours — advisor fires, human notified
        breach_after_seconds=14400, # 4 hours — instance marked STALLED
    ),
)

Query journey data in Cypher

Because everything is stored as graph nodes and edges, journey data is fully queryable:

-- Find all instances currently in the "classified" stage with SLA at risk
MATCH (i:JourneyInstance {current_stage: "classified", sla_status: "AT_RISK"})
RETURN i.instance_id, i.entity_id, i.entered_at
ORDER BY i.entered_at

-- Full audit trail for a specific instance
MATCH (i:JourneyInstance {instance_id: $id})-[t:ADVANCED_TO]->(stage)
RETURN t.from_stage, t.to_stage, t.actor, t.timestamp, t.notes
ORDER BY t.timestamp

-- AI decisions across all instances
MATCH ()-[a:AI_ADVISED]->()
WHERE a.action_type = "flag"
RETURN a.instance_id, a.reasoning, a.confidence, a.timestamp
ORDER BY a.timestamp DESC
LIMIT 20

Journey vs LangChain / CrewAI / AutoGen

Capability	LangChain / CrewAI / AutoGen	Purple8 Journey
Workflow state storage	External (Redis, Postgres, in-memory)	Graph-native — every transition = graph edge
AI decision context	Whatever you wire up	Full audit history + SLA health + metadata
Audit trail	DIY	✅ Immutable AI_ADVISED edges, append-only
Human-in-the-loop	Requires custom implementation	✅ auto_execute=False → human confirms
SLA enforcement	DIY	✅ SLAPolicy(warn_after_seconds, breach_after_seconds)
Real-time events	DIY	✅ EventBus → WebSocket CDC on every transition
Graph retrieval at decision time	Separate tool call	✅ Same engine — zero latency
Multi-tenant isolation	DIY	✅ Per-tenant JourneyEngine with JWT RBAC

---

See Also

• Customer Onboarding guide — end-to-end walkthrough: KYC, mainframe wiring, HITL compliance review, 360° customer queries
• Memory & Learning guide — using closed journey outcomes as LLM few-shot context
• MCP Integration guide — query any customer's journey in natural language via Claude or Cursor
• Real-time Augmented AI guide — combining CDC events with graph context