pyk.kore.rpc module

final class AbortedResult(state: 'State', depth: 'int', unknown_predicate: 'Pattern | None', logs: 'tuple[LogEntry, ...]', haskell_log_entries: 'tuple[Any, ...] | None' = None)

Bases: ExecuteResult

depth: int

classmethod from_dict(dct: Mapping[str, Any]) → AbortedResult

haskell_log_entries: tuple[Any, ...] | None = None

logs: tuple[LogEntry, ...]

next_states: tuple[State, ...] | None = None

reason: ClassVar[StopReason] = 'aborted'

rule: str | None = None

state: State

unknown_predicate: Pattern | None

class BoosterServer(args: BoosterServerArgs)

Bases: KoreServer

class BoosterServerArgs

Bases: dict

bug_report: BugReport | None

command: str | Iterable[str] | None

fallback_on: Iterable[str | FallbackReason] | None

haskell_log_entries: Iterable[str] | None

haskell_log_format: KoreExecLogFormat | None

haskell_threads: int | None

interim_simplification: int | None

kompiled_dir: Required[str | Path]

llvm_kompiled_dir: Required[str | Path]

log_axioms_file: Path | None

log_context: Iterable[str] | None

module_name: Required[str]

no_post_exec_simplify: bool | None

not_log_context: Iterable[str] | None

port: int | None

simplify_each: int | None

smt_reset_interval: int | None

smt_retry_limit: int | None

smt_tactic: str | None

smt_timeout: int | None

final class BranchingResult(state: 'State', depth: 'int', next_states: 'tuple[State, ...]', logs: 'tuple[LogEntry, ...]', haskell_log_entries: 'tuple[Any, ...] | None' = None)

Bases: ExecuteResult

depth: int

classmethod from_dict(dct: Mapping[str, Any]) → BranchingResult

haskell_log_entries: tuple[Any, ...] | None = None

logs: tuple[LogEntry, ...]

next_states: tuple[State, ...]

reason: ClassVar[StopReason] = 'branching'

rule: str | None = None

state: State

final class CutPointResult(state: 'State', depth: 'int', next_states: 'tuple[State, ...]', rule: 'str', logs: 'tuple[LogEntry, ...]', haskell_log_entries: 'tuple[Any, ...] | None' = None)

Bases: ExecuteResult

depth: int

classmethod from_dict(dct: Mapping[str, Any]) → CutPointResult

haskell_log_entries: tuple[Any, ...] | None = None

logs: tuple[LogEntry, ...]

next_states: tuple[State, ...]

reason: ClassVar[StopReason] = 'cut-point-rule'

rule: str

state: State

final exception DefaultError(message: 'str', code: 'int', data: 'Any' = None)

Bases: KoreClientError

code: int

data: Any

message: str

final class DepthBoundResult(state: 'State', depth: 'int', logs: 'tuple[LogEntry, ...]', haskell_log_entries: 'tuple[Any, ...] | None' = None)

Bases: ExecuteResult

depth: int

classmethod from_dict(dct: Mapping[str, Any]) → DepthBoundResult

haskell_log_entries: tuple[Any, ...] | None = None

logs: tuple[LogEntry, ...]

next_states: tuple[State, ...] | None = None

reason: ClassVar[StopReason] = 'depth-bound'

rule: str | None = None

state: State

final exception DuplicateModuleError(module_name: 'str')

Bases: KoreClientError

module_name: str

class ExecuteResult

Bases: ABC

depth: int

classmethod from_dict(dct: Mapping[str, Any]) → ER

haskell_log_entries: tuple[Any, ...] | None

logs: tuple[LogEntry, ...]

next_states: tuple[State, ...] | None

reason: ClassVar[StopReason]

rule: str | None

state: State

class FallbackReason(value)

Bases: Enum

An enumeration.

ABORTED = 'Aborted'

BRANCHING = 'Branching'

STUCK = 'Stuck'

class GetModelResult

Bases: ABC

static from_dict(dct: Mapping[str, Any]) → GetModelResult

haskell_log_entries: tuple[Any, ...] | None

final class HttpTransport(host: str, port: int, *, timeout: int | None = None)

Bases: Transport

close() → None

final exception ImplicationError(error: 'str', context: 'Iterable[str]')

Bases: KoreClientError

context: tuple[str, ...]

error: str

final class ImpliesResult(valid: 'bool', implication: 'Pattern', substitution: 'Pattern | None', predicate: 'Pattern | None', logs: 'tuple[LogEntry, ...]', haskell_log_entries: 'tuple[Any, ...] | None' = None)

Bases: object

static from_dict(dct: Mapping[str, Any]) → ImpliesResult

haskell_log_entries: tuple[Any, ...] | None = None

implication: Pattern

logs: tuple[LogEntry, ...]

predicate: Pattern | None

substitution: Pattern | None

valid: bool

final exception InvalidModuleError(error: 'str', context: 'Iterable[str] | None')

Bases: KoreClientError

context: tuple[str, ...] | None

error: str

class JsonRpcClient(host: str, port: int, *, timeout: int | None = None, bug_report: BugReport | None = None, bug_report_id: str | None = None, transport: TransportType = TransportType.SINGLE_SOCKET)

Bases: ContextManager[JsonRpcClient]

close() → None

interrupt() → None

property last_request_id: str | None

The JSON-RPC id of the most recent request issued by this client (None if none yet).

request(method: str, **params: Any) → dict[str, Any]

class JsonRpcClientFacade(default_host: str, default_port: int, default_transport: TransportType, dispatch: dict[str, list[tuple[str, int, TransportType]]], *, timeout: int | None = None, bug_report: BugReport | None = None, bug_report_id: str | None = None)

Bases: ContextManager[JsonRpcClientFacade]

close() → None

interrupt() → None

property last_request_id: str | None

The JSON-RPC id of the most recent request issued through this facade (None if none yet).

request(method: str, **params: Any) → dict[str, Any]

final exception JsonRpcError(message: 'str', code: 'int', data: 'Any' = None)

Bases: Exception

class KoreClient(host: str, port: int, *, timeout: int | None = None, bug_report: BugReport | None = None, bug_report_id: str | None = None, transport: TransportType = TransportType.SINGLE_SOCKET, dispatch: dict[str, list[tuple[str, int, TransportType]]] | None = None)

Bases: ContextManager[KoreClient]

add_module(module: Module, *, name_as_id: bool | None = None, haskell_logging: Iterable[str] | None = None) → str

close() → None

execute(pattern: Pattern, *, max_depth: int | None = None, assume_state_defined: bool | None = None, cut_point_rules: Iterable[str] | None = None, terminal_rules: Iterable[str] | None = None, moving_average_step_timeout: bool | None = None, step_timeout: int | None = None, module_name: str | None = None, log_successful_rewrites: bool | None = None, log_failed_rewrites: bool | None = None, booster_only_simplify: bool | None = None, haskell_logging: Iterable[str] | None = None) → ExecuteResult

get_model(pattern: Pattern, module_name: str | None = None, haskell_logging: Iterable[str] | None = None) → GetModelResult

implies(antecedent: Pattern, consequent: Pattern, *, module_name: str | None = None, assume_defined: bool = False, booster_only_simplify: bool | None = None, haskell_logging: Iterable[str] | None = None) → ImpliesResult

interrupt() → None

Abort an execute/simplify/… request running on another thread.

Sends a cancel so the server stops computing; the interrupted call raises a “cancelled” error and the connection stays usable. Works on the single-socket transport only; a no-op for HTTP (one connection per request, nothing to cancel).

property last_request_id: str | None

The JSON-RPC id of the most recent request issued by this client.

None before any request has been made. Snapshot it immediately after a call (e.g. execute/implies/simplify) to key per-request data (logs, handoff metadata) to that exact request. Each proof worker drives its own KoreClient (see _ProverPool), so this needs no cross-thread synchronization.

port: int

simplify(pattern: Pattern, *, module_name: str | None = None, booster_only_simplify: bool | None = None, haskell_logging: Iterable[str] | None = None) → SimplifyResult

exception KoreClientError(message: str)

Bases: Exception, ABC

class KoreExecLogFormat(value)

Bases: Enum

An enumeration.

JSON = 'json'

ONELINE = 'oneline'

STANDARD = 'standard'

class KoreServer(args: KoreServerArgs)

Bases: ContextManager[KoreServer]

close() → None

property host: str

property pid: int

property port: int

start() → None

class KoreServerArgs

Bases: TypedDict

bug_report: BugReport | None

command: str | Iterable[str] | None

haskell_log_entries: Iterable[str] | None

haskell_log_format: KoreExecLogFormat | None

haskell_threads: int | None

kompiled_dir: Required[str | Path]

log_axioms_file: Path | None

module_name: Required[str]

port: int | None

smt_reset_interval: int | None

smt_retry_limit: int | None

smt_tactic: str | None

smt_timeout: int | None

class KoreServerInfo(pid, host, port)

Bases: NamedTuple

host: str

Alias for field number 1

pid: int

Alias for field number 0

port: int

Alias for field number 2

class LogEntry

Bases: ABC

classmethod from_dict(dct: Mapping[str, Any]) → LE

abstract to_dict() → dict[str, Any]

class LogOrigin(value)

Bases: str, Enum

An enumeration.

BOOSTER = 'booster'

KORE_RPC = 'kore-rpc'

LLVM = 'llvm'

PROXY = 'proxy'

__format__(format_spec)

Returns format using actual value type unless __str__ has been overridden.

final class LogRewrite(origin: 'LogOrigin', result: 'RewriteResult')

Bases: LogEntry

classmethod from_dict(dct: Mapping[str, Any]) → LogRewrite

origin: LogOrigin

result: RewriteResult

to_dict() → dict[str, Any]

final exception ParseError(error: 'str')

Bases: KoreClientError

error: str

final exception PatternError(error: 'str', context: 'Iterable[str]')

Bases: KoreClientError

context: tuple[str, ...]

error: str

final class RewriteFailure(rule_id: 'str | None', reason: 'str')

Bases: RewriteResult

classmethod from_dict(dct: Mapping[str, Any]) → RewriteFailure

reason: str

rule_id: str | None

to_dict() → dict[str, Any]

class RewriteResult

Bases: ABC

classmethod from_dict(dct: Mapping[str, Any]) → RR

rule_id: str | None

abstract to_dict() → dict[str, Any]

final class RewriteSuccess(rule_id: 'str', rewritten_term: 'Pattern | None' = None)

Bases: RewriteResult

classmethod from_dict(dct: Mapping[str, Any]) → RewriteSuccess

rewritten_term: Pattern | None = None

rule_id: str

to_dict() → dict[str, Any]

final class SatResult(model: 'Pattern | None', haskell_log_entries: 'tuple[Any, ...] | None' = None)

Bases: GetModelResult

haskell_log_entries: tuple[Any, ...] | None = None

model: Pattern | None

class SimplifyResult(state: Pattern, logs: tuple[LogEntry, ...], haskell_log_entries: tuple[Any, ...] | None)

Bases: NamedTuple

Result of a simplify call.

A NamedTuple (not a bare tuple) so the optional haskell_log_entries bundle can be carried alongside the simplified state and logs without losing positional ergonomics.

haskell_log_entries: tuple[Any, ...] | None

Alias for field number 2

logs: tuple[LogEntry, ...]

Alias for field number 1

state: Pattern

Alias for field number 0

final class SingleSocketTransport(host: str, port: int, *, timeout: int | None = None)

Bases: Transport

close() → None

send_interrupt(data: str) → None

final exception SmtSolverError(error: 'str', pattern: 'Pattern')

Bases: KoreClientError

error: str

pattern: Pattern

final class State(term: 'Pattern', *, substitution: 'Mapping[EVar, Pattern] | None' = None, predicate: 'Pattern | None' = None, rule_id: 'str | None' = None, rule_substitution: 'Mapping[EVar, Pattern] | None' = None, rule_predicate: 'Pattern | None' = None)

Bases: object

static from_dict(dct: Mapping[str, Any]) → State

property kore: Pattern

predicate: Pattern | None

rule_id: str | None

rule_predicate: Pattern | None

rule_substitution: FrozenDict[EVar, Pattern] | None

substitution: FrozenDict[EVar, Pattern] | None

term: Pattern

class StopReason(value)

Bases: str, Enum

An enumeration.

ABORTED = 'aborted'

BRANCHING = 'branching'

CUT_POINT_RULE = 'cut-point-rule'

DEPTH_BOUND = 'depth-bound'

STUCK = 'stuck'

TERMINAL_RULE = 'terminal-rule'

TIMEOUT = 'timeout'

VACUOUS = 'vacuous'

__format__(format_spec)

Returns format using actual value type unless __str__ has been overridden.

final class StuckResult(state: 'State', depth: 'int', logs: 'tuple[LogEntry, ...]', haskell_log_entries: 'tuple[Any, ...] | None' = None)

Bases: ExecuteResult

depth: int

classmethod from_dict(dct: Mapping[str, Any]) → StuckResult

haskell_log_entries: tuple[Any, ...] | None = None

logs: tuple[LogEntry, ...]

next_states: tuple[State, ...] | None = None

reason: ClassVar[StopReason] = 'stuck'

rule: str | None = None

state: State

final class TerminalResult(state: 'State', depth: 'int', rule: 'str', logs: 'tuple[LogEntry, ...]', haskell_log_entries: 'tuple[Any, ...] | None' = None)

Bases: ExecuteResult

depth: int

classmethod from_dict(dct: Mapping[str, Any]) → TerminalResult

haskell_log_entries: tuple[Any, ...] | None = None

logs: tuple[LogEntry, ...]

next_states: tuple[State, ...] | None = None

reason: ClassVar[StopReason] = 'terminal-rule'

rule: str

state: State

final class TimeoutResult(state: 'State', depth: 'int', logs: 'tuple[LogEntry, ...]', haskell_log_entries: 'tuple[Any, ...] | None' = None)

Bases: ExecuteResult

depth: int

classmethod from_dict(dct: Mapping[str, Any]) → TimeoutResult

haskell_log_entries: tuple[Any, ...] | None = None

logs: tuple[LogEntry, ...]

next_states: tuple[State, ...] | None = None

reason: ClassVar[StopReason] = 'timeout'

rule: str | None = None

state: State

class Transport

Bases: ContextManager[Transport], ABC

abstract close() → None

request(req: str, request_id: str, method_name: str) → str

send_interrupt(data: str) → None

Send data on the live connection without waiting for a reply.

Used to deliver a cancel to a connection whose reply another thread is already awaiting. Default: no-op. Override only for connections that can be written to while a request is in flight.

class TransportType(value)

Bases: Enum

An enumeration.

HTTP = 2

SINGLE_SOCKET = 1

final exception UnknownModuleError(module_name: 'str')

Bases: KoreClientError

module_name: str

final class UnknownResult(haskell_log_entries: 'tuple[Any, ...] | None' = None)

Bases: GetModelResult

haskell_log_entries: tuple[Any, ...] | None = None

final class UnsatResult(haskell_log_entries: 'tuple[Any, ...] | None' = None)

Bases: GetModelResult

haskell_log_entries: tuple[Any, ...] | None = None

final class VacuousResult(state: 'State', depth: 'int', logs: 'tuple[LogEntry, ...]', haskell_log_entries: 'tuple[Any, ...] | None' = None)

Bases: ExecuteResult

depth: int

classmethod from_dict(dct: Mapping[str, Any]) → VacuousResult

haskell_log_entries: tuple[Any, ...] | None = None

logs: tuple[LogEntry, ...]

next_states: tuple[State, ...] | None = None

reason: ClassVar[StopReason] = 'vacuous'

rule: str | None = None

state: State

kore_server(definition_dir: str | Path, module_name: str, *, port: int | None = None, command: str | Iterable[str] | None = None, smt_timeout: int | None = None, smt_retry_limit: int | None = None, smt_tactic: str | None = None, log_axioms_file: Path | None = None, haskell_log_format: KoreExecLogFormat | None = None, haskell_log_entries: Iterable[str] | None = None, haskell_threads: int | None = None, llvm_definition_dir: Path | None = None, fallback_on: Iterable[str | FallbackReason] | None = None, interim_simplification: int | None = None, simplify_each: int | None = None, no_post_exec_simplify: bool | None = None, bug_report: BugReport | None = None) → KoreServer