primus.cli.procedure.tactus_adapters package

Primus adapters for Tactus protocols.

These adapters implement Tactus protocols to integrate the standalone Tactus library with Primus’s GraphQL-based infrastructure.

class primus.cli.procedure.tactus_adapters.OptimizerExternalChildResolver(*, backend: Any, account_id: str)

Bases: object

Resolve exact, persisted optimizer children without account scans.

A malformed/missing/mismatched record is an incomplete child, never a substitute identity and never a success. Failed or cancelled children are terminal only after both indexed records corroborate the same immutable launch identity.

__init__(*, backend: Any, account_id: str) → None
class primus.cli.procedure.tactus_adapters.PrimusChatAdapter(chat_recorder)

Bases: object

Implements Tactus ChatRecorder protocol by wrapping ProcedureChatRecorder.

This is a thin adapter that converts Pydantic ChatMessage models to the kwargs format expected by ProcedureChatRecorder.

Initialize Primus chat adapter.

Args:

chat_recorder: ProcedureChatRecorder instance

__init__(chat_recorder)

Initialize Primus chat adapter.

Args:

chat_recorder: ProcedureChatRecorder instance

async end_session(session_id: str, status: str = 'COMPLETED') → None

End the chat session.

Args:

session_id: Session ID to end status: Final status (COMPLETED, FAILED, CANCELLED)

async get_session_history(session_id: str) → list[ChatMessage]

Get the message history for a session.

Args:

session_id: Session ID

Returns:

List of ChatMessage objects

async record_message(message: ChatMessage) → str

Record a message in the chat session.

Args:

message: ChatMessage to record

Returns:

Message ID

async start_session(context: Dict[str, Any] | None = None) → str

Start a new chat session.

Note: This signature matches how TactusRuntime actually calls it (runtime.py:214), which only passes context. The Tactus ChatRecorder protocol definition incorrectly specifies procedure_id as the first parameter, but this is not how it’s used.

Args:

context: Optional context data

Returns:

Session ID

class primus.cli.procedure.tactus_adapters.PrimusHITLAdapter(client, procedure_id: str, chat_recorder=None, storage_adapter=None)

Bases: object

Tactus HITLHandler implementation backed by Primus ChatMessage records.

__init__(client, procedure_id: str, chat_recorder=None, storage_adapter=None)
cancel_pending_request(procedure_id: str, message_id: str) → None
check_pending_response(procedure_id: str, message_id: str, *, request: HITLRequest | None = None) → HITLResponse | None
request_interaction(procedure_id: str, request: HITLRequest, execution_context: Any = None) → HITLResponse
class primus.cli.procedure.tactus_adapters.PrimusStorageAdapter(client, procedure_id: str, *, artifact_store=None)

Bases: object

Implements Tactus StorageBackend protocol using Primus GraphQL.

Stores all procedure data (checkpoints, state, lua_state) in the Procedure.metadata JSON field via GraphQL mutations.

Initialize Primus storage adapter.

Args:

client: PrimusDashboardClient instance procedure_id: ID of the procedure

__init__(client, procedure_id: str, *, artifact_store=None)

Initialize Primus storage adapter.

Args:

client: PrimusDashboardClient instance procedure_id: ID of the procedure

checkpoint_clear_after(procedure_id: str, name: str) → None

Clear checkpoint and all subsequent ones.

checkpoint_clear_all(procedure_id: str) → None

Clear all checkpoints (but preserve state).

checkpoint_exists(procedure_id: str, name: str) → bool

Check if checkpoint exists.

checkpoint_get(procedure_id: str, name: str) → Any | None

Get checkpoint value.

checkpoint_save(procedure_id: str, name: str, result: Any) → None

Save a checkpoint.

get_state(procedure_id: str) → Dict[str, Any]

Get mutable state dictionary.

load_procedure_metadata(procedure_id: str) → ProcedureMetadata

Load verified procedure state and checkpoints through artifact tickets.

save_procedure_metadata(procedure_id: str, metadata: ProcedureMetadata) → None

Save procedure metadata to Primus via GraphQL.

Args:

procedure_id: Procedure ID (for API compatibility) metadata: ProcedureMetadata to save

set_state(procedure_id: str, state: Dict[str, Any]) → None

Set mutable state dictionary.

state_clear(procedure_id: str) → None

Clear all state.

state_delete(procedure_id: str, key: str) → None

Delete state key.

state_get(procedure_id: str, key: str, default: Any = None) → Any

Get state value.

state_set(procedure_id: str, key: str, value: Any) → None

Set state value.

update_procedure_status(procedure_id: str, status: str, waiting_on_message_id: str | None = None) → None

Update procedure status in Primus.

Args:

procedure_id: Procedure ID status: New status waiting_on_message_id: Optional message ID if waiting for human

class primus.cli.procedure.tactus_adapters.PrimusTraceSink(chat_recorder)

Bases: object

Persist Tactus trace records into Primus ChatSession/ChatMessage models.

STREAM_UPDATE_MAX_INTERVAL_SECONDS = 0.25
STREAM_UPDATE_MIN_CHARS_DELTA = 16
__init__(chat_recorder)
async end_session(status: str = 'COMPLETED') → None
async flush() → None
mark_runtime_execute_started(value: str | None = None) → None
async record(event: Any) → str | None
async start_session(context: Dict[str, Any] | None = None) → str | None
class primus.cli.procedure.tactus_adapters.TerminalHITLAdapter(auto_approve: bool = False)

Bases: object

Tactus HITLHandler that resolves HITL requests via terminal prompts.

Used when running procedures from the CLI so that Human.approve() and Human.input() calls block at the terminal rather than suspending the procedure and waiting for a dashboard response.

Args:

auto_approve: If True, automatically approve all requests without prompting.

__init__(auto_approve: bool = False)
Args:

auto_approve: If True, automatically approve all requests without prompting.

request_interaction(procedure_id: str, request: HITLRequest, execution_context: Any = None) → HITLResponse

Submodules