praval.hitl.store

SQLite-backed persistence for HITL interventions and suspended runs.

Functions

get_hitl_store([db_path])

Get a shared HITL store instance by path.

reset_hitl_stores()

Reset shared store cache (primarily for tests).

Classes

HITLStore([db_path])

Durable storage for intervention queues and paused runs.
class praval.hitl.store.HITLStore(db_path=None)[source]

Bases: object

Durable storage for intervention queues and paused runs.

Parameters:

db_path (Optional[str])

__init__(db_path=None)[source]
Parameters:

db_path (str | None)

create_intervention(*, run_id, agent_name, provider_name, tool_name, tool_call_id, original_args, risk_level='low', approval_reason='', trace_id=None, metadata=None, expires_in_seconds=86400)[source]

Create a new pending intervention.

Return type:

InterventionRequest

Parameters:

  • run_id (str)

  • agent_name (str)

  • provider_name (str)

  • tool_name (str)

  • tool_call_id (str)

  • original_args (Dict[str, Any])

  • risk_level (str)

  • approval_reason (str)

  • trace_id (str | None)

  • metadata (Dict[str, Any] | None)

  • expires_in_seconds (int)

get_intervention(intervention_id)[source]

Fetch a single intervention by id.

Return type:

Optional[InterventionRequest]

Parameters:

intervention_id (str)

list_interventions(*, status=None, run_id=None, agent_name=None, limit=100)[source]

List interventions with optional filters.

Return type:

List[InterventionRequest]

Parameters:

  • status (InterventionStatus | None)

  • run_id (str | None)

  • agent_name (str | None)

  • limit (int)

list_pending_interventions(*, run_id=None, agent_name=None, limit=100)[source]

List pending interventions with optional filters.

Return type:

List[InterventionRequest]

Parameters:

  • run_id (str | None)

  • agent_name (str | None)

  • limit (int)

decide_intervention(intervention_id, *, decision, reviewer, reason='', edited_args=None)[source]

Apply a human decision for a pending intervention.

Return type:

InterventionRequest

Parameters:

  • intervention_id (str)

  • decision (InterventionDecision)

  • reviewer (str)

  • reason (str)

  • edited_args (Dict[str, Any] | None)

expire_overdue_interventions()[source]

Mark expired pending interventions as EXPIRED.

Return type:

int

upsert_suspended_run(*, run_id, agent_name, provider_name, state, status='pending')[source]

Create or update suspended run state.

Return type:

SuspendedRunState

Parameters:

  • run_id (str)

  • agent_name (str)

  • provider_name (str)

  • state (Dict[str, Any])

  • status (str)

get_suspended_run(run_id)[source]

Get suspended run state by run_id.

Return type:

Optional[SuspendedRunState]

Parameters:

run_id (str)

update_suspended_run_status(run_id, status, state=None)[source]

Update run status and optional state payload.

Return type:

None

Parameters:

  • run_id (str)

  • status (str)

  • state (Dict[str, Any] | None)

transition_suspended_run(run_id, *, from_status, to_status)[source]

Move a run from from_status to to_status atomically.

Return type:

bool

Returns:

False when the run does not exist or is no longer in from_status, for example because another thread or process claimed it first.

Parameters:

  • run_id (str)

  • from_status (str)

  • to_status (str)

update_suspended_run_state(run_id, state, *, expected_status)[source]

Replace a run’s state only while it is in expected_status.

Return type:

bool

Returns:

False when the run does not exist or has moved to another status, for example because a new intervention re-pended it.

Parameters:

  • run_id (str)

  • state (Dict[str, Any])

  • expected_status (str)

delete_suspended_run(run_id)[source]

Delete suspended run state.

Return type:

None

Parameters:

run_id (str)

praval.hitl.store.get_hitl_store(db_path=None)[source]

Get a shared HITL store instance by path.

Return type:

HITLStore

Parameters:

db_path (str | None)

praval.hitl.store.reset_hitl_stores()[source]

Reset shared store cache (primarily for tests).

Return type:

None