inspection
Shared runtime inspection and JSON serialization helpers.
attributelogger= logging.getLogger(__name__)funcdescribe_local_server() -> strSummarize the state of the local Kitaru-compatible server, if any.
Returns
strfuncis_registered_local_server_url(url) -> boolReturn whether a URL matches the registered local Kitaru server.
paramurlstr | NoneReturns
boolfuncconnected_to_local_server_safe() -> boolSafely check whether the current client is bound to a local server.
Returns
boolfuncuses_stale_local_server_url(server_url, local_server_status) -> boolCheck for a localhost URL that points at a stopped local server.
paramserver_urlstr | Noneparamlocal_server_statusstr | NoneReturns
boolfunclog_store_mismatch_details(preferred) -> tuple[str | None, str | None]Return status-row + warning text when preferred and active backends differ.
parampreferredResolvedLogStoreReturns
tuple[str | None, str | None]funccombine_warnings(*warnings) -> str | NoneCombine non-empty warning messages into one multiline block.
paramwarningsstr | None= ()Returns
str | Nonefuncbuild_runtime_snapshot(*, include_packages=False, package_names=None, include_environment_type=False) -> RuntimeSnapshotResolve the current Kitaru runtime state from ZenML-backed config.
paraminclude_packagesbool= Falseparampackage_namesSequence[str] | None= Noneparaminclude_environment_typebool= FalseReturns
kitaru.inspection.RuntimeSnapshotfuncto_jsonable(value, *, fallback_repr) -> AnyConvert a value into a JSON-serializable representation.
paramvalueAnyparamfallback_reprboolReturns
typing.Anyfuncserialize_failure(failure) -> dict[str, Any] | NoneSerialize optional failure details.
paramfailureFailureInfo | NoneReturns
dict[str, typing.Any] | Nonefuncserialize_pending_wait(wait) -> dict[str, Any] | NoneSerialize optional pending wait details.
paramwaitPendingWait | NoneReturns
dict[str, typing.Any] | Nonefuncserialize_artifact_ref(artifact) -> dict[str, Any]Serialize artifact metadata.
paramartifactArtifactRefReturns
dict[str, typing.Any]funcserialize_artifact_value(value) -> dict[str, Any]Serialize an artifact payload value for MCP transport.
paramvalueAnyReturns
dict[str, typing.Any]funcserialize_memory_entry(entry) -> dict[str, Any]Serialize typed memory-entry metadata for transport layers.
paramentryMemoryEntryReturns
dict[str, typing.Any]funcserialize_memory_history(history) -> list[dict[str, Any]]Serialize a memory history sequence.
paramhistorySequence[MemoryEntry]Returns
list[dict[str, typing.Any]]funcserialize_memory_scope_info(info) -> dict[str, Any]Serialize a discovered memory scope for transport layers.
paraminfoMemoryScopeInfoReturns
dict[str, typing.Any]funcserialize_memory_value(value) -> dict[str, Any]Serialize a loaded memory value using the shared artifact-value rules.
paramvalueAnyReturns
dict[str, typing.Any]funcserialize_purge_result(result) -> dict[str, Any]Serialize a purge result for transport layers.
paramresultPurgeResultReturns
dict[str, typing.Any]funcserialize_compaction_record(record) -> dict[str, Any]Serialize a compaction audit record for transport layers.
paramrecordCompactionRecordReturns
dict[str, typing.Any]funcserialize_compact_result(result) -> dict[str, Any]Serialize a compact result for transport layers.
paramresultCompactResultReturns
dict[str, typing.Any]funcserialize_memory_reindex_issue(issue) -> dict[str, Any]Serialize a sampled memory reindex issue for transport layers.
paramissueMemoryReindexIssueReturns
dict[str, typing.Any]funcserialize_memory_reindex_result(result) -> dict[str, Any]Serialize a memory reindex result for transport layers.
paramresultMemoryReindexResultReturns
dict[str, typing.Any]funcserialize_checkpoint_attempt(attempt) -> dict[str, Any]Serialize checkpoint-attempt details.
paramattemptCheckpointAttemptReturns
dict[str, typing.Any]funcserialize_checkpoint_call(checkpoint) -> dict[str, Any]Serialize checkpoint-call details.
paramcheckpointCheckpointCallReturns
dict[str, typing.Any]funcserialize_execution_summary(execution) -> dict[str, Any]Serialize execution list-item details.
paramexecutionExecutionReturns
dict[str, typing.Any]funcserialize_execution(execution) -> dict[str, Any]Serialize full execution details.
paramexecutionExecutionReturns
dict[str, typing.Any]funcserialize_stack(stack, *, is_managed=None) -> dict[str, Any]Serialize stack information for structured output.
paramstackStackInfoparamis_managedbool | None= NoneReturns
dict[str, typing.Any]funcserialize_stack_create_result(result) -> dict[str, Any]Serialize stack-create operation details.
paramresult_StackCreateResultReturns
dict[str, typing.Any]funcserialize_stack_delete_result(result) -> dict[str, Any]Serialize stack-delete operation details.
paramresult_StackDeleteResultReturns
dict[str, typing.Any]funcserialize_stack_details(details) -> dict[str, Any]Serialize stack inspection details for stack show style output.
paramdetailsStackDetailsReturns
dict[str, typing.Any]funcserialize_runtime_snapshot(snapshot) -> dict[str, Any]Serialize runtime status details for structured output.
paramsnapshotRuntimeSnapshotReturns
dict[str, typing.Any]funcserialize_log_entry(entry) -> dict[str, Any]Serialize one log entry for JSON output.
paramentryLogEntryReturns
dict[str, typing.Any]funcserialize_model_alias(entry) -> dict[str, Any]Serialize model alias information.
paramentryModelAliasEntryReturns
dict[str, typing.Any]funcsecret_visibility(secret) -> strReturn a human-readable visibility label for a secret.
paramsecretSecretResponseReturns
strfuncserialize_secret_summary(secret) -> dict[str, Any]Serialize secret summary information.
paramsecretSecretResponseReturns
dict[str, typing.Any]funcserialize_secret_detail(secret, *, show_values) -> dict[str, Any]Serialize secret detail information.
paramsecretSecretResponseparamshow_valuesboolReturns
dict[str, typing.Any]funcserialize_resolved_log_store(snapshot, *, active_store=None, warning=None) -> dict[str, Any]Serialize effective log-store information.
paramsnapshotResolvedLogStoreparamactive_storeActiveStackLogStore | None= Noneparamwarningstr | None= NoneReturns
dict[str, typing.Any]