Core API Reference

This page is generated from public docstrings in praxist.core.

Registry

registry

Run-local plugin discovery, resolution, and registry support.

PluginRef dataclass

Parsed kind:name plugin reference with optional source, version, and requiredness constraints.

PluginMetadata dataclass

Canonical plugin manifest data used for resolution, hashing, loading, and replay.

PluginIdentity dataclass

Minimal identity tuple used during discovery before full metadata validation.

PluginCandidate dataclass

Discovered plugin instance from one bundled, project, user, or task_project root.

SelectedPlugin dataclass

Resolved plugin descriptor with source, path, content hash, and selection provenance.

DiscoveryReport dataclass

Plugin discovery output, including candidates and non-fatal manifest warnings.

PluginRoots dataclass

Ordered plugin root set used by discovery for bundled, project, user, and task sources.

defaults classmethod

defaults(
    workspace: Path | None = None,
    task_path: Path | None = None,
    *,
    extra_bundled: list[Path] | None = None,
    extra_project: list[Path] | None = None,
) -> PluginRoots

Build the default plugin-root search set.

extra_bundled / extra_project (issue #75 batch 6) let CLI entrypoints inject additional plugin search paths explicitly, bypassing the env-fallback. When either argument is None the helper falls back to PRAXIST_BUNDLED_PLUGIN_ROOTS / PRAXIST_PLUGIN_ROOTS so the test fixture path (tests/__init__.py mutates PRAXIST_BUNDLED_PLUGIN_ROOTS to inject fixture plugins) and operator-level env extension both continue to work. Passing an empty list ([]) explicitly disables the env-fallback — useful for hermetic builds.

PluginRegistry

Frozen registry of selected plugin descriptors and optional instantiated objects.

PluginRegistryBuilder

Mutable registry builder used only during manifest loading.

PluginLoader

Thin Gate B loader: discover, resolve, then freeze selected descriptors.

selected_plugin_from_dict

selected_plugin_from_dict(
    value: dict[str, Any],
) -> SelectedPlugin

Rehydrate a SelectedPlugin from a resolution manifest record.

require_execution_plugin

require_execution_plugin(
    registry: PluginRegistry | None,
    ref: str,
    *,
    kind: str,
    capability: str | None = None,
) -> SelectedPlugin | None

Validate that a selected plugin is executable under the current source policy.

compute_plugin_content_hash

compute_plugin_content_hash(
    plugin_path: Path,
    metadata: PluginMetadata | dict[str, Any] | None = None,
) -> str

Hash a plugin manifest plus declared code and assets under its root.

read_plugin_metadata

read_plugin_metadata(plugin_path: Path) -> PluginMetadata

Read canonical plugin metadata from the on-disk manifest.

static_selected_plugin

static_selected_plugin(
    ref: PluginRef, selected_by: list[str]
) -> SelectedPlugin

Create a deterministic selected-plugin descriptor for legacy static manifests.

static_resolution_manifest

static_resolution_manifest(
    run_id: str, root_task_ref: str, refs: list[str]
) -> dict[str, Any]

Build a static plugin resolution manifest for compatibility-only paths.

assert_bundled_execution_manifest

assert_bundled_execution_manifest(
    manifest: dict[str, Any],
) -> None

Verify executable plugin selections come from a trusted source (bundled or task_project) until dispatch is registry-backed.

load_plugin_entrypoint

load_plugin_entrypoint(selected: SelectedPlugin) -> Any

Load a selected plugin's manifest-declared entrypoint factory.

instantiate_plugin_entrypoint

instantiate_plugin_entrypoint(
    selected: SelectedPlugin,
) -> Any

Instantiate a selected executable plugin using its entrypoint factory.

dumps_manifest

dumps_manifest(value: dict[str, Any]) -> str

Serialize a manifest with stable ordering for human inspection and tests.

Task Projects

task_project

Task project loading for explicit, external research tasks.

Task projects are not bundled Praxist plugins. They are user-owned project directories selected with --task-path and may contain task-local roles, audits, evaluations, harness assets, and optional reference implementations.

TaskProject dataclass

Resolved task project selected by an explicit filesystem path.

resolve_task_project

resolve_task_project(
    task_path: str | Path,
    workspace: str | Path | None = None,
) -> TaskProject

Load and fingerprint a task project directory.

task_path may point at either the project directory or its task.yaml. Relative paths are resolved against workspace when provided.

task_project_has_capability

task_project_has_capability(
    project: TaskProject, capability: str
) -> bool

Return whether a resolved task project declares a given capability.

task_project_global_plugin_refs

task_project_global_plugin_refs(
    descriptor: dict[str, Any],
) -> list[PluginRef]

Return bundled plugin refs required by a task descriptor.

Task-local refs such as task_role:peer remain inside the task project and are intentionally omitted from bundled plugin resolution.

is_task_local_ref

is_task_local_ref(ref: str | None) -> bool

Return whether a reference is scoped to the explicit task project rather than global plugins.

build_task_project_manifest

build_task_project_manifest(
    root: Path,
    *,
    descriptor_path: Path | None = None,
    descriptor: dict[str, Any] | None = None,
) -> dict[str, Any]

Hash a task project descriptor and source files into a replayable manifest.

load_task_project_runner

load_task_project_runner(project: TaskProject) -> Any

Import the task-local runner factory declared by a task project descriptor.

write_task_project_manifest

write_task_project_manifest(
    run_dir: str | Path, project: TaskProject
) -> Path

Persist a resolved task project manifest into the run directory.

Protocol

protocol

Gate B protocol dataclasses shared by core, runtimes, and providers.

SANDBOX_FILESYSTEM_VALUES module-attribute

SANDBOX_FILESYSTEM_VALUES: tuple[str, ...] = (
    "read_only",
    "workspace_write",
    "full",
)

Allowed values for :attr:RuntimeSandboxIntent.filesystem.

SANDBOX_NETWORK_VALUES module-attribute

SANDBOX_NETWORK_VALUES: tuple[str, ...] = ('off', 'on')

Allowed values for :attr:RuntimeSandboxIntent.network.

SANDBOX_APPROVAL_VALUES module-attribute

SANDBOX_APPROVAL_VALUES: tuple[str, ...] = (
    "auto",
    "on_risk",
    "always_ask",
)

Allowed values for :attr:RuntimeSandboxIntent.approval.

SANDBOX_ENFORCEMENT_APPROVAL_GATE module-attribute

SANDBOX_ENFORCEMENT_APPROVAL_GATE = 'approval_gate'

Runtime enforces sandbox intent through interactive approval prompts only.

SANDBOX_ENFORCEMENT_OS_SANDBOX module-attribute

SANDBOX_ENFORCEMENT_OS_SANDBOX = 'os_sandbox'

Runtime enforces sandbox intent through OS-level isolation (seatbelt, landlock, ...).

ModelProfile dataclass

Declarative model capability profile selected by task, role, stage, or budget policy.

ModelCallSpec dataclass

Provider-ready model call configuration derived from a ModelProfile and provider adapter.

ModelResult dataclass

Normalized provider response metadata used for runtime accounting and failure classification.

ToolPermissionSet dataclass

Tool allowance contract passed from workflow stages to agent runtime adapters.

ToolServerRef dataclass

Resolved tool-server endpoint visible to an agent or panel role.

ToolCallResult dataclass

Normalized result of an in-process or MCP-shaped tool invocation.

EnvPolicy dataclass

Scoped environment injection policy for runtime, tool, and subprocess execution.

CachePolicy dataclass

Runtime/provider cache strategy recorded for prompt-layout replay checks.

AgentRunRequest dataclass

Serializable request passed from workflow stages to an AgentRuntime adapter.

AgentEvent dataclass

Normalized stream event emitted by an AgentRuntime for trajectory and replay.

ToolCallRecord dataclass

Compact record of one runtime-observed tool call.

AgentRunResult dataclass

Normalized terminal result of one agent runtime execution.

BudgetRequest dataclass

Serializable request for compute, wall-clock, token, data, or tool budget.

BudgetGrant dataclass

Approved budget envelope that execution guards can enforce and meter.

BudgetDecision dataclass

Policy output describing grant, deny, downscope, defer, or require-review decisions.

RuntimeSandboxIntent dataclass

Runtime-neutral sandbox intent declared by the operator or task project.

Captures what the operator wants the agent runtime to allow. Each :class:AgentRuntime plugin declares in its manifest which values of each axis it can honor; resolution fails fast when the intent contains a value the chosen runtime cannot enforce.

The vocabulary is intentionally small: three coarse axes that map cleanly to most CLI agents' approval and sandbox flags. Finer-grained needs are escape-hatched via per-runtime raw flags rather than growing this enum.

Attributes:

Name	Type	Description
filesystem	FilesystemIntent	Filesystem write scope intent. "read_only" means the runtime should reject writes outside its own state; "workspace_write" permits writes inside the active run directory; "full" lets the runtime touch anywhere the process user can write.
network	NetworkIntent	Network egress intent. "off" requests OS-level network isolation; "on" permits outbound network calls.
approval	ApprovalIntent	Action-approval policy intent. "auto" lets the runtime act without prompting; "on_risk" prompts before mutating operations; "always_ask" prompts before every action.

to_dict

to_dict() -> dict[str, str]

Return a JSON-serializable view for trajectory and replay.

Runtimes

runtimes

AgentRuntime registry-backed loader helpers.

AgentRuntimeExecutionContext dataclass

Process-local handles shared by asynchronous AgentRuntime plugins.

AgentRunRequest remains the complete serializable execution contract. This context carries only values that cannot safely or usefully be put in protocol JSON: instantiated tool servers, callbacks, stop polling, and the already-scoped environment prepared by the workflow stage.

RuntimeUsageCollector dataclass

Aggregate runtime usage for the current workflow-stage context.

effective_reasoning_effort

effective_reasoning_effort(
    runtime_options: Mapping[str, Any] | None,
) -> str

Resolve the task-level reasoning policy without assuming a provider API.

premium_mode remains a compatibility alias for max when the new policy is absent or auto. Runtime adapters own the provider-specific wire mapping.

collect_runtime_usage

collect_runtime_usage() -> Iterator[RuntimeUsageCollector]

Collect usage from runtime executions in this asynchronous context.

runtime_for_ref

runtime_for_ref(
    runtime_ref: str, registry: PluginRegistry | None = None
) -> Any

Resolve and instantiate an AgentRuntime implementation for a plugin reference.

runtime_managed_credential_for_ref

runtime_managed_credential_for_ref(
    runtime_ref: str,
    model_provider_ref: str,
    registry: PluginRegistry,
) -> credentials.CredentialRef | None

Ask a runtime for optional non-environment authentication.

This is an optional plugin extension. Core validates the returned redacted reference but remains independent of any runtime's authentication backend.

resolve_model_credential_for_runtime

resolve_model_credential_for_runtime(
    credential_set: CredentialSet,
    runtime_ref: str,
    model_provider_ref: str,
    registry: PluginRegistry,
    *,
    resolve_only: bool,
) -> tuple[
    credentials.CredentialSet,
    credentials.CredentialRef | None,
]

Resolve env credentials before an optional runtime-managed fallback.

execute_runtime async

execute_runtime(
    runtime: Any,
    request: AgentRunRequest,
    context: AgentRuntimeExecutionContext,
) -> AgentRunResult

Execute one runtime through the common async contract.

Production runtimes implement execute. Deterministic fixture plugins may keep the smaller synchronous execute_sync contract used by conformance tests.

prompt_text_for_request

prompt_text_for_request(request: AgentRunRequest) -> str

Return the normalized inline user prompt from an agent request.

system_prompt_text_for_request

system_prompt_text_for_request(
    request: AgentRunRequest,
) -> str | None

Return the process-local system prompt carried by runtime options.

classify_runtime_failure

classify_runtime_failure(
    error: str | None, *, timed_out: bool = False
) -> str

Classify a redacted runtime error into the shared failover vocabulary.

is_provider_access_error

is_provider_access_error(error: str) -> bool

Return whether a runtime failure reflects credentials or account quota.

close_runtime_for_ref async

close_runtime_for_ref(
    runtime_ref: str, registry: PluginRegistry | None
) -> None

Close a runtime's process-local resources when its workflow stage ends.

event_types_for_conformance

event_types_for_conformance(
    runtime_refs: list[str], request: AgentRunRequest
) -> dict[str, list[str]]

Return the normalized runtime event types expected by conformance tests.

Modeling

modeling

ModelProfile helpers and registry-backed ModelProvider dispatch.

ModelProviderAdapter

Small provider contract used by core before a full provider SDK call is made.

validate_model_for_provider

validate_model_for_provider(
    provider_ref: str,
    model: str,
    registry: PluginRegistry | None = None,
) -> None

Validate that a concrete model name is compatible with the selected provider adapter.

normalize_model_for_provider

normalize_model_for_provider(
    provider_ref: str,
    model: str,
    registry: PluginRegistry | None = None,
) -> str

Reshape a model name into the format the provider's api_format expects.

Openrouter uses vendor/model; the other api_formats we ship (openai_compatible, anthropic_messages, fake) consume bare model names. Operator configs and env vars often carry the vendor/ prefix as a leftover from the era when openrouter was the only supported access path; this helper strips that prefix when the resolved provider is not openrouter so the same configuration works against either an aggregator or a direct provider endpoint.

provider_for_ref

provider_for_ref(
    provider_ref: str,
    registry: PluginRegistry | None = None,
) -> ModelProviderAdapter

Return the model provider adapter for a registry plugin reference.

default_model_profile

default_model_profile(
    provider_ref: str,
    *,
    profile_id: str = "cheap_peer",
    cost_tier: str = "cheap",
    model: str | None = None,
    registry: PluginRegistry | None = None,
) -> ModelProfile

Create the default ModelProfile selected for the current model provider and model name.

model_profiles_snapshot

model_profiles_snapshot(
    *,
    provider_ref: str,
    runtime_ref: str,
    credential_mode: str,
    cache_policy: Any,
    selected_model: str | None = None,
    registry: PluginRegistry | None = None,
) -> dict[str, Any]

Build the run-local snapshot written to model_profiles.json for replay and cost attribution.

provider_default_model

provider_default_model(
    provider_ref: str,
    registry: PluginRegistry | None = None,
) -> str | None

Return the plugin yaml's default_model for provider_ref.

Single source-of-truth lookup for "what model does this provider pick when the operator did not supply one?". Returns None when the plugin cannot be loaded or has no default_model declared; callers should fall back to a built-in literal (e.g. :data:praxist.core.run_config.DEFAULT_AGENT_MODEL).

Centralising this here lets the yaml be the only place an operator has to edit to change a provider's default — no parallel hardcoded chain in workflow-stage startup code (see #144).

Budget

budget

Budget policy registry dispatch and shared unit validation.

BudgetPolicy

Bases: Protocol

Protocol for deterministic budget policies that turn requests into grants, denials, or review decisions.

policy_for_ref

policy_for_ref(
    policy_ref: str, registry: PluginRegistry | None = None
) -> BudgetPolicy

Return the bundled or registry-backed BudgetPolicy implementation for a plugin reference.

Credentials

credentials

CredentialRef skeleton and redacted env discovery.

CredentialRef dataclass

Stable, redacted reference to one provider credential without storing the raw secret.

CredentialSet dataclass

Resolved credential inventory for one provider, including robust-mode fallback order.

CredentialResolver

Discover credentials without exposing raw secret values.

CredentialFailoverManager

Minimal robust-mode selector with per-key failure state.

provider_name_from_ref

provider_name_from_ref(model_provider_ref: str) -> str

Extract the provider name component from a model-provider plugin reference.

find_model_provider_credential

find_model_provider_credential(
    credential_set: CredentialSet, model_provider_ref: str
) -> CredentialRef | None

Find the best available credential set for a selected model provider.

require_model_provider_credential

require_model_provider_credential(
    credential_set: CredentialSet, model_provider_ref: str
) -> CredentialRef | None

Resolve a provider credential or fail fast with an operator-facing error.

Prompt Layout

prompt_layout

PromptLayout V1 helpers.

Step 20 keeps existing legacy prompt text runnable while making the cache-relevant prompt boundaries explicit. Jinja remains a renderer for individual blocks; it does not own the frozen/dynamic partition.

PromptBlock dataclass

Typed prompt segment with cache stability metadata and a deterministic content hash.

PromptLayout dataclass

Rendered prompt plus block-level hashes used for cache readiness and replay verification.

build_legacy_jinja_prompt_layout

build_legacy_jinja_prompt_layout(
    *,
    base_template_path: Path,
    task_prompt_path: Path | None,
    generation_template_path: Path | None,
    context: dict[str, Any],
    run_id: str,
    stage_id: str,
    prompt_id: str,
    agent_runtime_ref: str = "agent_runtime:claude_sdk",
    model_provider_ref: str = "",
    repo_root: Path | None = None,
    frozen_prefix_text: str = DEFAULT_FROZEN_PREFIX,
    start_command_text: str = DEFAULT_START_COMMAND,
    extra_dynamic_blocks: list[dict[str, Any]]
    | None = None,
) -> PromptLayout

Render the legacy research-loop prompt as PromptLayout V1 blocks.

Existing Jinja templates remain dynamic legacy blocks. A stable prefix is prepended so runtime auto-cache implementations have an invariant prefix to reuse while Step 20 separates the remaining legacy monolith.

write_prompt_layout_files

write_prompt_layout_files(
    *,
    layout: PromptLayout,
    prompt_path: Path,
    manifest_path: Path,
    rendered_prompt_ref: dict[str, Any] | None = None,
) -> dict[str, Any]

Persist prompt text and layout metadata artifacts for one agent session.

audit_frozen_blocks

audit_frozen_blocks(
    blocks: list[PromptBlock],
) -> dict[str, Any]

Validate that frozen prompt blocks do not contain dynamic run, peer, or generation markers.

find_dynamic_markers

find_dynamic_markers(text: str) -> list[str]

Return dynamic marker names detected in prompt text.

sha256_text

sha256_text(text: str) -> str

Return a SHA-256 digest for UTF-8 prompt text.

sha256_json

sha256_json(value: Any) -> str

Return a stable SHA-256 digest for a JSON-serializable object.

Tool Servers

tool_servers

Core ToolServer selection, construction, and result normalization.

Concrete MCP factories and handlers live under praxist/plugins/tools/*. This module owns registry-backed selection, mode gates, allowed-tool naming, and normalized tool-call accounting.

ToolServerSpec dataclass

Resolved tool-server plugin contract used for MCP registration and tool-name permissions.

ToolServerBuildResult dataclass

Result of building in-process legacy MCP servers for a workflow stage.

tool_server_refs_from_task_descriptor

tool_server_refs_from_task_descriptor(
    descriptor: dict[str, Any],
) -> tuple[str, ...]

Extract runtime-visible tool-server references from a task descriptor.

effective_research_tool_server_refs_from_task_descriptor

effective_research_tool_server_refs_from_task_descriptor(
    descriptor: dict[str, Any],
) -> tuple[str, ...]

Return the research-loop tool-server refs Praxist will actually use.

The research loop has long used DEFAULT_RESEARCH_TOOL_SERVER_REFS when a task descriptor does not declare tool servers. Startup plugin resolution must use the same effective selection so resolve-only cannot pass with a registry that later lacks the default tool descriptors.

tool_server_for_ref

tool_server_for_ref(
    ref: str, registry: PluginRegistry | None = None
) -> ToolServerSpec

Resolve one tool_server plugin reference into a ToolServerSpec.

build_legacy_mcp_servers

build_legacy_mcp_servers(
    tool_refs: Iterable[str] | None = None,
    *,
    run_dir: Path | str | None = None,
    local_mode: bool,
    multi_pi_enabled: bool = False,
    registry: PluginRegistry | None = None,
) -> ToolServerBuildResult

Instantiate selected legacy in-process MCP servers for a run.

allowed_mcp_tool_names

allowed_mcp_tool_names(
    tool_refs: Iterable[str] | None = None,
    *,
    local_mode: bool,
    include_panel_tools: bool = False,
    include_peer_tools: bool = True,
    multi_pi_enabled: bool = False,
    registry: PluginRegistry | None = None,
) -> list[str]

Return fully qualified MCP tool names allowed for a peer or panel role.

allowed_mcp_tool_names_for_servers

allowed_mcp_tool_names_for_servers(
    server_names: Iterable[str],
    *,
    include_panel_tools: bool = False,
    include_peer_tools: bool = True,
    tool_refs: Iterable[str] | None = None,
    registry: PluginRegistry | None = None,
) -> list[str]

Return allowed MCP tool names from already connected server names.

visible_mcp_servers

visible_mcp_servers(
    servers: dict[str, Any],
    *,
    include_panel_tools: bool = False,
    include_peer_tools: bool = True,
    tool_refs: Iterable[str] | None = None,
    registry: PluginRegistry | None = None,
) -> dict[str, Any]

Filter connected MCP servers by role visibility.

Runtime adapters receive server descriptors separately from tool permissions. A panel-only server must therefore be removed from peer requests, not merely omitted from the allowed-tool list.

base_peer_allowed_tools

base_peer_allowed_tools(
    server_names: Iterable[str],
    *,
    include_panel_tools: bool = False,
    tool_refs: Iterable[str] | None = None,
    registry: PluginRegistry | None = None,
) -> list[str]

Return built-in Claude Code tools plus allowed MCP tools for a peer.

peer_mcp_context

peer_mcp_context(
    servers: dict[str, Any],
    *,
    tool_refs: Iterable[str] | None = None,
    registry: PluginRegistry | None = None,
) -> tuple[dict[str, Any], list[str]]

Return peer-visible MCP servers plus matching allowed tool names.

execute_legacy_tool_handler_async async

execute_legacy_tool_handler_async(
    server_ref: str,
    tool_name: str,
    args: dict[str, Any] | None = None,
    *,
    registry: PluginRegistry | None = None,
    run_dir: Path | str | None = None,
    run_id: str = "",
    budget_grant_id: str | None = None,
    budget_request_id: str | None = None,
    stage_id: str = "research_loop",
) -> ToolCallResult

Execute a manifest-declared legacy tool handler with budget and redaction accounting.

execute_legacy_tool_handler

execute_legacy_tool_handler(
    server_ref: str,
    tool_name: str,
    args: dict[str, Any] | None = None,
    *,
    registry: PluginRegistry | None = None,
    run_dir: Path | str | None = None,
    run_id: str = "",
    budget_grant_id: str | None = None,
    budget_request_id: str | None = None,
    stage_id: str = "research_loop",
) -> ToolCallResult

Synchronous wrapper for executing a legacy tool handler.

normalize_tool_result

normalize_tool_result(
    server_name: str, tool_name: str, raw: Any
) -> ToolCallResult

Convert raw MCP-style handler output into a ToolCallResult.

Role Skills

role_skills

RoleSkill loading for declarative role plugins.

RoleSkill dataclass

Loaded role prompt contract, including skill markdown, tool scope, private KB, and content hash.

load_role_skill

load_role_skill(
    role_ref: str,
    *,
    registry: PluginRegistry | None = None,
    workspace: Path | None = None,
    task_project_path: Path | None = None,
) -> RoleSkill

Load a bundled role plugin or task-local role skill for prompt assembly.

Workflow

workflow

Core workflow stage protocol helpers and optional-stage stubs.

WorkflowStageSpec dataclass

Declarative workflow stage contract resolved from plugin metadata.

WorkflowStageResult dataclass

Terminal result returned by a workflow stage entrypoint.

OptionalWorkflowStageContext dataclass

Serializable description of an optional stage that may be disabled at startup.

OptionalWorkflowStageStub

Disabled-by-default contract stub for future optional workflow modules.

disabled_optional_stages

disabled_optional_stages() -> list[dict[str, str | bool]]

Return optional workflow stages disabled by a task descriptor.

disabled_optional_tools

disabled_optional_tools() -> list[dict[str, str | bool]]

Return optional tool refs disabled by a task descriptor.

optional_workflow_stage

optional_workflow_stage(
    stage_id: str,
) -> OptionalWorkflowStageStub

Create the disabled/enabled contract for a named optional workflow stage.

literature_scout_contract

literature_scout_contract() -> dict[str, Any]

Return the optional task-local literature_scout role/tool contract.

emit_disabled_optional_events

emit_disabled_optional_events(
    trajectory: Any,
    *,
    stages: list[dict[str, Any]] | None = None,
    tools: list[dict[str, Any]] | None = None,
    research_stage_id: str = "research_loop",
) -> None

Record disabled optional stages and tools into trajectory for replay visibility.

Storage

storage

Run directory, JSONL, and artifact helpers for Gate A.

ArtifactWriter

Run-local artifact writer with stable ids, redaction, and artifact_index accounting.

utc_now

utc_now() -> str

Return the current UTC timestamp used by Praxist run artifacts.

utc_stamp

utc_stamp() -> str

Return a filesystem-safe UTC timestamp string.

new_run_id

new_run_id(task_slug: str) -> str

Generate a new opaque run identifier.

write_json

write_json(path: Path, value: Any) -> None

Write a redacted JSON artifact with stable indentation and sorted keys.

append_jsonl

append_jsonl(path: Path, value: Any) -> None

Append one redacted JSONL record to an append-only run ledger.

read_jsonl

read_jsonl(
    path: Path,
) -> tuple[list[dict[str, Any]], list[str]]

Read JSONL records from a run ledger, skipping blank lines.

rewrite_jsonl

rewrite_jsonl(
    path: Path, records: list[dict[str, Any]]
) -> None

Atomically rewrite an entire JSONL ledger with redacted records.

sha256_bytes

sha256_bytes(data: bytes) -> str

Return a SHA-256 digest for raw bytes.

output_ledger_hashes

output_ledger_hashes(run_dir: Path) -> dict[str, str]

Compute content hashes for canonical output ledgers in a run directory.

ensure_run_dirs

ensure_run_dirs(run_dir: Path) -> None

Create the minimum run directory layout used by startup and replay.

Trajectory

trajectory

Append-only trajectory writer for Gate A.

TrajectoryWriter

Append-only trajectory writer that assigns stable event ids and redacts persisted fields.

Replay

replay

Inspect / verify / dry-run support for run directories.

Replay is an internal explainability and consistency mechanism for fast Praxist runs. The default verifier should help humans and downstream modules understand what happened after the fact; locked mode is reserved for benchmark/release artifacts that need strict drift failures.

inspect_run

inspect_run(run_dir: Path) -> dict[str, Any]

Inspect a run directory and return counts, artifacts, and replay-facing diagnostics.

verify_run

verify_run(
    run_dir: Path,
    *,
    strict_tail: bool = False,
    allow_plugin_drift: bool = False,
    locked: bool = False,
) -> dict[str, Any]

Verify replay invariants for run lifecycle, redaction, plugin provenance, usage, and materialized state.

dry_run

dry_run(
    run_dir: Path,
    *,
    strict_tail: bool = False,
    allow_plugin_drift: bool = False,
    locked: bool = False,
) -> dict[str, Any]

Return a non-mutating replay plan that summarizes what verify would inspect.

Execution Guards

execution_guards

Budget and observability helpers for execution resource guards.

ResourceBudgetError

Bases: RuntimeError

Raised before a high-cost action starts without an approved grant.

BudgetedActionReport dataclass

Lifecycle report for one budget-guarded runtime, tool, GPU, or evaluation action.

BudgetedActionGuard

Record per-action usage without making observability a result killer.

from_run_config classmethod

from_run_config(
    run_config: RunConfig,
    *,
    action_type: str,
    actor_ref: str,
    require_budget_grant: bool = False,
    metadata: dict[str, Any] | None = None,
) -> BudgetedActionGuard

Build a guard from an explicit :class:RunConfig (issue #75 batch 4).

Sibling of :meth:from_env for callers that have already constructed a RunConfig at the CLI boundary. Preserves the same defaults (stage_id="research_loop", run_id derived from run_dir.name when explicit run_id is empty, budget_grant_id / request_id None when empty) so the two constructors produce equivalent guards from equivalent inputs.

record_budgeted_action_from_env

record_budgeted_action_from_env(
    *,
    action_type: str,
    actor_ref: str,
    actual_usage: dict[str, float] | None = None,
    expected_units: list[str] | tuple[str, ...] = (),
    status: str = "succeeded",
    reason: str = "action_usage",
    metadata: dict[str, Any] | None = None,
) -> BudgetedActionReport

Record action usage from PRAXIST_* environment variables used by legacy subprocesses.

emit_resource_event_from_env

emit_resource_event_from_env(
    kind: str,
    *,
    action_type: str,
    actor_ref: str,
    payload: dict[str, Any] | None = None,
    severity: str = "info",
) -> None

Emit a resource lifecycle trajectory event from the current scoped execution environment.

record_budgeted_action_from_run_config

record_budgeted_action_from_run_config(
    run_config: RunConfig,
    *,
    action_type: str,
    actor_ref: str,
    actual_usage: dict[str, float] | None = None,
    expected_units: list[str] | tuple[str, ...] = (),
    status: str = "succeeded",
    reason: str = "action_usage",
    metadata: dict[str, Any] | None = None,
) -> BudgetedActionReport

record_budgeted_action_from_env sibling that takes a :class:RunConfig (issue #75 batch 4).

emit_resource_event_from_run_config

emit_resource_event_from_run_config(
    run_config: RunConfig,
    kind: str,
    *,
    action_type: str,
    actor_ref: str,
    payload: dict[str, Any] | None = None,
    severity: str = "info",
) -> None

emit_resource_event_from_env sibling that takes a :class:RunConfig (issue #75 batch 4).

gpu_hours_since

gpu_hours_since(
    started_at: str, *, finished_at: datetime | None = None
) -> float | None

Convert elapsed wall-clock time and GPU count into gpu_hours accounting units.

Source Snapshot

source_snapshot

Source snapshot helpers for replay drift detection.

build_core_source_snapshot

build_core_source_snapshot(
    repo_root: Path | None = None,
) -> dict[str, Any]

Hash execution-critical Praxist source files for replay drift detection.