Sessions and persistence

QitOS public API: sessions

Obtain Session through composition.session/restore/fork or Engine.session; do not call its constructor directly. run(steering=…) executes and returns EngineResult; inspect reads a SessionInspection. pause requests a cooperative boundary and returns a receipt, not proof of immediate termination. Session identities, run identities and snapshot/checkpoint identities are distinct. SQLite supports clean-process recovery; Memory is process-local. Unresolved approvals and CLI live pause/steer remain unsupported.

完整可运行教程 / Complete tutorial · API index

Signatures and fields below are extracted from the pinned source. Signatures are reference material, not standalone programs. Source links bind the same runtime baseline. Any does not imply arbitrary objects are supported; use the behavioral contract above and the linked tutorial.

Session

from qitos.engine.session_runtime import Session

Source @ 19f6258

Usage and executable example

Usage fragment: continues with objects from the linked complete tutorial; not a standalone program.

session = composition.session("Index notes")
result = session.run()
inspection = session.inspect()
print(session.session_id.value, result.state.final_result)

Scoped client for one durable Session identity.

The facade stores identifiers and cooperative control only. Agent state is
reconstructed from the canonical snapshot and executed by ``Engine.run``.

Session(*, engine: 'Engine[Any, Any, Any]', session_id: SessionIdentity, run_id: RunIdentity, agent_id: AgentIdentity, references: Iterable[ResolverReference], created_at: str, state_type: type[StateSchema], work_item_id: WorkItemIdentity, attempt_id: AttemptIdentity, fork_receipt: Optional[SessionForkReceipt]=None) -> None

Parameter	Type	Default
engine	'Engine[Any, Any, Any]'	required
session_id	SessionIdentity	required
run_id	RunIdentity	required
agent_id	AgentIdentity	required
references	Iterable[ResolverReference]	required
created_at	str	required
state_type	type[StateSchema]	required
work_item_id	WorkItemIdentity	required
attempt_id	AttemptIdentity	required
fork_receipt	Optional[SessionForkReceipt]	None

Session.run

run(*, steering: Optional[str]=None) -> 'EngineResult[Any]'

Parameter	Type	Default
steering	Optional[str]	None

Run or resume through the one canonical Engine loop.

Source

Session.inspect

inspect() -> SessionInspection

Source

Session.pause

pause() -> PauseReceipt

Request cooperative pause; durable status is returned at a boundary.

Source

Session.steer

steer(text: str) -> SteeringReceipt

Parameter	Type	Default
text	str	required

Durably submit one canonical steering item to this Session.

Source

Session.fork

fork(snapshot: SessionSnapshot | SnapshotIdentity | str | None=None, *, operation_id: Optional[str]=None) -> 'Session'

Parameter	Type	Default
snapshot	SessionSnapshot | SnapshotIdentity | str | None	None
operation_id	Optional[str]	None

Create an isolated durable child from one verified immutable snapshot.

Source

Session.capabilities

capabilities() -> frozenset[str]

Source

SessionInspection

from qitos.engine.session_runtime import SessionInspection

Source @ 19f6258

Usage and executable example

Usage fragment: continues with objects from the linked complete tutorial; not a standalone program.

inspection = session.inspect()
print(inspection.work_graph)

Read-only inspection result backed by the current durable head.

Field	Type	Default
head	SessionHead	required
lifecycle	SessionLifecycle	required
capabilities	tuple[str, ...]	required
snapshot_integrity	str	required
budget	Mapping[str, Any]	required
work_graph	Optional[Mapping[str, Any]]	required
last_request_view	Optional[RequestView]	required
task	str	required
tool_batch	Optional[ToolBatchSnapshot]	required

CheckpointStore

from qitos.checkpoint.store import CheckpointStore

Source @ 19f6258

Usage and executable example

Usage fragment: continues with objects from the linked complete tutorial; not a standalone program.

# From an open composition configured with SQLite:
store = composition.runtime.checkpoint_store
head = store.get_session_head(session.session_id.value)
print(head)

Abstract base class for checkpoint persistence.

Borrowed from LangGraph's ``BaseCheckpointSaver`` interface
(``references/langgraph/libs/checkpoint/langgraph/checkpoint/base/__init__.py``).

Every method has both sync and async variants.  Subclasses should
override the async variants; the sync ones delegate via ``asyncio.run``
by default.

CheckpointStore.get_session_head

get_session_head(session_id: str) -> Optional[SessionHeadRecord]

Parameter	Type	Default
session_id	str	required

Read the current mutable head for one session.

Source

CheckpointStore.commit_session_snapshot

commit_session_snapshot(request: SessionSnapshotCommit) -> SessionCommitReceipt

Parameter	Type	Default
request	SessionSnapshotCommit	required

Atomically persist an immutable snapshot and advance its head.

Source