Sessions and operations - Nebius Token Factory documentation

Start with Work with a Session, Save and resume a session, or Advanced: control a running sandbox. Sync calls return results directly. Async calls must be awaited unless used as the PendingRun context described in the operation guide. ContreeSession and ContreeAsyncSession save each command by default. disposable=True leaves the session image unchanged. LazySession and AsyncLazySession reuse a VM until snapshot. Snapshots are lightweight and can run after every command, but they stop the VM. An unchanged filesystem can return the same image UUID. See Save after every command.

`class AbstractCommitPolicy`

class contree_sdk.session.AbstractCommitPolicy()

Decide whether a validated operation result advances session history. Sessions call this synchronous method only after API success, for a non-disposable request. Implementations must not perform blocking I/O. A policy cannot accept API failure, cancellation, or a missing result image.

Methods

`should_commit()`

@abstractmethod
def should_commit(context, result) -> bool

required

bool

`class AbstractSnapshotPolicy`

class contree_sdk.session.AbstractSnapshotPolicy()

Own the decision state and expose one future per VM cycle. Use one policy instance per lazy session. Subclasses call super().init() and super().notify(event) to reset or cancel the cycle future. Resolve the future to request a snapshot, or fail it to report a policy error. The session still waits for all accepted commands and inspections before stopping its VM.

Methods

`notify()`

@abstractmethod
def notify(event) -> None

Handle an execution event without blocking or performing network I/O.

required

None

`should_snapshot()`

def should_snapshot() -> Future[None]

Return this cycle’s future; successful completion requests a snapshot.The caller must not cancel or resolve it. To await it from asyncio, use shield(wrap_future(…)) so cancellation does not affect the policy.

Future[None]

A future shared by all readers of the current policy decision.

`class ApiSuccessCommitPolicy`

class contree_sdk.session.ApiSuccessCommitPolicy()

Retain every successful API result, including a nonzero or unknown exit code.

Methods

`should_commit()`

def should_commit(context, result) -> bool

required

bool

`class AsyncLazySession`

class contree_sdk.session.AsyncLazySession(session, *, snapshot_policy, snapshot_timeout, vm_timeout)

Share one VM across commands; save one history entry per snapshot. The supplied session, its client and its store remain caller-owned. Do not mutate that session outside this wrapper while a VM is active. Use this object as a context manager or call close/abort to release its worker.

Attributes

BaseException | None

Literal['cold', 'starting', 'running', 'draining', 'snapshotting', 'failed', 'closed']

str

Methods

`abort()`

async def abort() -> None

Discard unsaved work and close. An already running snapshot finishes first.

None

`close()`

async def close(*, timeout) -> None

Save pending work and close this wrapper, retaining caller-owned resources.

default:"None"

None

RuntimeError – Saving failed; the cause is available as error.

TimeoutError – The caller’s wait expired; closing continues in the worker.

`create_branch()`

async def create_branch(name, *, from_branch) -> None

required

default:"None"

None

`execute()`

async def execute(request) -> InstanceResult

required

InstanceResult

`grep()`

async def grep(pattern, *, path, glob, max_count, max_total, case, before, after) -> GrepResult

Search session files using the current source’s syntax and limits.

required

default:"None"

GrepResult

Matching lines and the server’s truncation indicator.

`inspect()`

async def inspect(*, timeout) -> AsyncSessionInspector

Return a view that selects the current filesystem for each request.Live requests do not stop the VM. Cold requests use the saved image; pending staged files start a VM so the view includes those files. timeout limits admission and VM startup waits, not HTTP I/O. Use inspect_snapshot() for a stable image and its image_uuid.

default:"None"

AsyncSessionInspector

A session-bound inspector; creating it starts no VM.

`inspect_snapshot()`

async def inspect_snapshot(*, timeout) -> AsyncImageInspector

Save active work and pin the resulting image for inspection.Waits for admitted commands and inspections, then stops the VM. A cold session starts no VM. Later commands cannot change the pinned image. timeout limits the snapshot wait; saving continues if the wait ends early. A failed snapshot raises instead of returning the previous image. Staged attachments appear only after a command applies them.

default:"None"

AsyncImageInspector

An inspector borrowing the underlying client and file transfer.

`inspection_source()`

def inspection_source(*, timeout) -> AsyncIterator[AsyncImageInspector | str]

Reserve the current source until the request and its stream close.Override this hook to customize source selection. Image inspectors are pinned; operation UUIDs remain protected from automatic snapshot.

default:"None"

AsyncIterator[AsyncImageInspector | str]

A saved image inspector or an active operation UUID.

RuntimeError – The session fails or closes before inspection starts.

TimeoutError – Admission or VM startup exceeds the timeout.

`keepalive_request()`

def keepalive_request() -> RunRequest

Supply a long-running main process available in the base image.

RunRequest

RunRequestThe non-disposable request used to start each VM.

`list_dir()`

async def list_dir(path) -> DirectoryList

List session directory entries relative to the filesystem root.

default:"'/'"

DirectoryList

Directory entries and their filesystem metadata.

`observe()`

async def observe(event, error) -> None

Observe the operation’s sole reader without depending on handle.wait().

required

None

`prepare_request()`

def prepare_request(request) -> RunRequest

Apply the wrapped session’s policy and validate shared-VM semantics.

required

RunRequest

RunRequestThe effective request.

ValueError – The request needs a disposable VM or startup-only options.

`read_file()`

async def read_file(path) -> bytes

Read current files without stopping an active VM.

required

bytes

The complete file as bytes.

`rollback()`

async def rollback(steps) -> None

default:"1"

None

`run()`

async def run(command, **kwargs) -> InstanceResult

Run a shared-VM command. Options are RunRequest fields; disposable defaults to False.

default:"None"

InstanceResult

The subprocess result, including a nonzero exit code.

`snapshot()`

async def snapshot(*, timeout) -> HistoryEntry | None

Drain commands and inspections, save their common image, and reopen admission.Snapshots are lightweight and can run after every command. They stop the VM. An unchanged filesystem can return the same image UUID. timeout limits this wait; an already requested snapshot continues. Errors remain available as error and are raised on subsequent calls. Saving failure or a closed wrapper raises RuntimeError. A wait that expires raises TimeoutError; saving continues in the worker.

default:"None"

HistoryEntry | None

HistoryEntry | NoneThe saved history entry, or None if no VM has run yet.

`spawn()`

async def spawn(command, **kwargs) -> AsyncSubprocessContract

default:"None"

AsyncSubprocessContract

AsyncSubprocessContract

`spawn_request()`

async def spawn_request(request) -> AsyncSubprocessContract

Start a subprocess; track its exit even if the caller never waits.

required

AsyncSubprocessContract

AsyncSubprocessContractThe subprocess handle from the operation factory.

`switch_branch()`

async def switch_branch(name) -> None

required

None

`write_file()`

async def write_file(path, content) -> None

Write bytes with the same contract as SyncExecutor.write_file.The default awaits execute() with binary stdin. Override with native async file transfer, coordinating with commands and snapshot lifecycle. Returns after application, not after staging or blob upload. Invalid paths raise ValueError; a failed write raises FileWriteError. Transport errors, timeouts, and cancellation propagate unchanged.

required

None

`class AsyncOperation`

class contree_sdk.session.AsyncOperation(client, operation_uuid, *, timeout, shutdown_timeout, files, context)

Handle to a spawned instance operation, identified by its UUID.

Attributes

str

OperationContext | None

OperationContext | None

OperationResponse | None

Methods

`add_observer()`

def add_observer(observer) -> None

Register an observer before starting the shared operation reader.A None event marks EOF, with an optional stream error. Observers must not consume events or wait for this operation. Implementations without this optional facility reject registration.

required

None

`apply_files()`

async def apply_files(files) -> None

Apply already-uploaded files to this running VM.

required

None

`cancel()`

async def cancel() -> None

None

`claim_queue()`

async def claim_queue(spid) -> Queue[OperationEvent | None]

required

Queue[OperationEvent | None]

`create_subprocess()`

def create_subprocess(spid, events_queue) -> AsyncSubprocessContract

Create a subprocess handle. Override to return a custom implementation.

required

AsyncSubprocessContract

AsyncSubprocessContractA handle consuming the supplied event queue.

`events()`

async def events(*, since, spid, timeout) -> AsyncIterator[OperationEvent]

Replay buffered events, then follow the shared reader with local filters.timeout bounds this subscription. Expiry does not cancel the operation.

default:"None"

AsyncIterator[OperationEvent]

Events after since whose process matches spid, when specified.

TimeoutError – The subscription did not finish before timeout.

`open_event_stream()`

def open_event_stream() -> AsyncIterator[OperationEvent]

Open the unfiltered source. Override this hook to supply custom events.

AsyncIterator[OperationEvent]

The single transport stream consumed by the operation.

`pipe_stdin()`

async def pipe_stdin(chunks, *, spid, close, chunk_size) -> StdinResult

Forward chunks after spawn and stop waiting for input when the process exits.Use one writer per process. Errors and cancellation stop the target process; spid=1 cancels the operation. A pending source read is cancelled on exit; the caller owns the source.

required

default:"1"

default:"True"

default:"65536"

StdinResult

StdinResultAcknowledged bytes, EOF delivery, and observed early completion.

ValueError – chunk_size or spid is not positive.

`pump()`

async def pump() -> None

None

`route_event()`

async def route_event(event) -> None

required

None

`run()`

async def run(command, *, shell, args, env, cwd, uid, gid, stdin, stdin_open, truncate_output_at) -> AsyncSubprocessContract

required

default:"False"

default:"()"

default:"None"

default:"False"

default:"None"

AsyncSubprocessContract

AsyncSubprocessContract

`send_stdin()`

async def send_stdin(data, *, spid, close) -> None

required

default:"1"

default:"True"

None

`shutdown()`

async def shutdown() -> None

None

`signal()`

async def signal(sig, *, spid) -> None

default:"None"

default:"1"

None

`status()`

async def status(*, inflight) -> OperationResponse

default:"False"

OperationResponse

`wait()`

async def wait(*, timeout) -> InstanceResult

default:"None"

InstanceResult

`class AsyncOperationContract`

class contree_sdk.session.AsyncOperationContract()

An operation lifecycle. wait stores response before returning; cleanup preserves primary errors.

Attributes

str

OperationContext | None

OperationContext | None

OperationResponse | None

Methods

`add_observer()`

def add_observer(observer) -> None

required

None

`apply_files()`

def apply_files(files) -> None

required

None

`cancel()`

def cancel() -> None

None

`events()`

@abstractmethod
def events(*, since, spid, timeout) -> AsyncIterator[OperationEvent]

default:"None"

AsyncIterator[OperationEvent]

`pipe_stdin()`

def pipe_stdin(chunks, *, spid, close, chunk_size) -> StdinResult

required

default:"1"

default:"True"

default:"65536"

StdinResult

StdinResult

`run()`

def run(command, *, shell, args, env, cwd, uid, gid, stdin, stdin_open, truncate_output_at) -> AsyncSubprocessContract

required

default:"False"

default:"()"

default:"None"

default:"False"

default:"None"

AsyncSubprocessContract

AsyncSubprocessContract

`send_stdin()`

def send_stdin(data, *, spid, close) -> None

required

default:"1"

default:"True"

None

`shutdown()`

def shutdown() -> None

None

`signal()`

def signal(sig, *, spid) -> None

default:"None"

default:"1"

None

`status()`

def status(*, inflight) -> OperationResponse

default:"False"

OperationResponse

`wait()`

def wait(*, timeout) -> InstanceResult

default:"None"

InstanceResult

`class AsyncSubprocessContract`

class contree_sdk.session.AsyncSubprocessContract()

Waitable event stream for one subprocess. Iteration and wait share completion state.

Attributes

int

Methods

`pipe_to()`

def pipe_to(*, stdout, stderr) -> InstanceResult

default:"None"

InstanceResult

`wait()`

def wait(*, timeout) -> InstanceResult

default:"None"

InstanceResult

`class AsyncSubprocessHandle`

class contree_sdk.session.AsyncSubprocessHandle(operation, spid, events_queue)

Both awaitable (final result) and async-iterable (live events) for one spid.

Attributes

int

Methods

`iterate()`

async def iterate() -> AsyncIterator[OperationEvent]

AsyncIterator[OperationEvent]

`pipe_to()`

async def pipe_to(*, stdout, stderr) -> InstanceResult

default:"None"

InstanceResult

`wait()`

async def wait(*, timeout) -> InstanceResult

default:"None"

InstanceResult

`class CommandCountSnapshotPolicy`

class contree_sdk.session.CommandCountSnapshotPolicy(limit)

Request a snapshot after exactly limit command admissions in one VM.

Methods

`notify()`

def notify(event) -> None

Handle an execution event without blocking or performing network I/O.

required

None

`should_snapshot()`

def should_snapshot() -> Future[None]

Future[None]

A future shared by all readers of the current policy decision.

`class CompositeSnapshotPolicy`

class contree_sdk.session.CompositeSnapshotPolicy(*policies)

Request a snapshot when any child policy’s future completes.

Methods

`notify()`

def notify(event) -> None

Handle an execution event without blocking or performing network I/O.

required

None

`should_snapshot()`

def should_snapshot() -> Future[None]

Future[None]

A future shared by all readers of the current policy decision.

`class ContreeAsyncSession`

class contree_sdk.session.ContreeAsyncSession(client, *, image, session_id, node_id, tree_kind, store, file_transfer, cwd, env, commit_policy)

A durable, resumable ConTree session backed by a Store.

Attributes

str

Methods

`check_operation_endpoint()`

def check_operation_endpoint(record) -> None

Check the recorded server before using an operation UUID.

required

None

ValueError – The current client targets a different endpoint.

`commit_result()`

async def commit_result(operation, *, title, branch, files, data) -> HistoryEntry | None

required

default:"None"

HistoryEntry | None

HistoryEntry | None

`create_branch()`

async def create_branch(name, *, from_branch) -> None

required

default:"None"

None

`create_file_transfer()`

def create_file_transfer() -> AsyncFileTransfer

Create file I/O when no component was supplied.

AsyncFileTransfer

AsyncFileTransferA file-transfer component using the session client.

`create_inspector()`

def create_inspector(image_uuid) -> AsyncImageInspector

Create a view of one image. Override to customize inspection.

required

AsyncImageInspector

An inspector borrowing the client’s connection and file transfer.

`create_operation()`

def create_operation(response, context) -> AsyncOperationContract

Construct a lifecycle implementation bound to the supplied operation context.

required

AsyncOperationContract

AsyncOperationContractAn implementation of the matching operation lifecycle contract.

`create_operation_record()`

def create_operation_record(operation) -> OperationRecord

Serialize an operation context. Override to persist custom request fields.

required

OperationRecord

OperationRecordA pending record without local input sources.

ValueError – The operation does not belong to this session.

`create_pending_run()`

def create_pending_run(request, *, branch) -> PendingRun

Create the awaitable/context-manager wrapper for a command.

required

default:"None"

PendingRun

PendingRunA one-use wrapper for awaiting or entering the operation.

`create_store()`

def create_store() -> AsyncStore

Create the default store when none was supplied.

AsyncStore

AsyncStoreA new memory store owned by this session.

`delete_branch()`

async def delete_branch(name) -> None

required

None

`detach()`

async def detach(operation) -> OperationRecord

required

OperationRecord

OperationRecordIts durable record. The caller still owns the live handle.

ValueError – The operation does not belong to this session.

`ensure_ready()`

async def ensure_ready() -> None

None

`execute()`

async def execute(request, *, branch) -> InstanceResult

Run a request to completion and commit its effective non-disposable result.

required

default:"None"

InstanceResult

The completed process result, including nonzero exit codes.

`get_entry()`

async def get_entry(node_id) -> HistoryEntry

required

HistoryEntry

HistoryEntry

`get_operation()`

async def get_operation(operation_uuid) -> OperationRecord

required

OperationRecord

OperationRecord

`grep()`

async def grep(pattern, *, path, glob, max_count, max_total, case, before, after) -> GrepResult

Search session files using the current source’s syntax and limits.

required

default:"None"

GrepResult

Matching lines and the server’s truncation indicator.

`history()`

async def history() -> tuple[list[HistoryEntry], dict[str, list[str]]]

tuple[list[HistoryEntry], dict[str, list[str]]]

tuple[list[HistoryEntry], dict[str, list[str]]]

`inspect()`

async def inspect() -> AsyncImageInspector

Pin the current saved image without starting a VM.Pending staged attachments appear only after an operation applies them.

AsyncImageInspector

A read-only view that borrows this session’s client and file transfer.

`list_branches()`

async def list_branches() -> list[tuple[str, bool]]

list[tuple[str, bool]]

`list_dir()`

async def list_dir(path) -> DirectoryList

List session directory entries relative to the filesystem root.

default:"'/'"

DirectoryList

Directory entries and their filesystem metadata.

`list_operations()`

async def list_operations(*, pending_only) -> tuple[OperationRecord, ...]

default:"True"

tuple[OperationRecord, ...]

tuple[OperationRecord, …]

`navigate()`

async def navigate(node_id) -> None

required

None

`navigate_forward()`

async def navigate_forward(steps) -> None

default:"1"

None

`pending_files()`

async def pending_files(node_id) -> tuple[StagedFile, ...]

default:"None"

tuple[StagedFile, ...]

tuple[StagedFile, …]

`prepare_environment()`

def prepare_environment(env) -> dict[str, str] | None

Overlay command variables on session defaults without mutating either input.

required

dict[str, str] | None

A fresh mapping, or None when both inputs omit environment variables.

`prepare_request()`

def prepare_request(request) -> RunRequest

Apply defaults before uploads or execution. Override to validate or transform commands.

required

RunRequest

RunRequestThe effective command with session defaults applied.

`read_file()`

async def read_file(path) -> bytes

required

bytes

`refresh_from_entry()`

async def refresh_from_entry(entry) -> None

required

None

`resolve_image()`

async def resolve_image(image) -> str

Resolve a base image. Override to implement an import or registry policy.

required

str

The resolved image UUID.

`restore_operation()`

async def restore_operation(operation_uuid) -> AsyncOperationContract

Recreate a control handle from the persisted source context.

required

AsyncOperationContract

AsyncOperationContractAn unentered handle. Use wait_operation to reuse a stored final result.

`restore_operation_context()`

def restore_operation_context(record) -> OperationContext

Decode a stored request. Override together with create_operation_record.

required

OperationContext

OperationContextIts immutable source context, ready for an operation factory or policy.

`rollback()`

async def rollback(steps) -> None

default:"1"

None

`run()`

def run(command, *, shell, args, env, cwd, stdin, stdin_open, files, timeout, disposable, truncate_output_at, preserve_env, hostname, uid, gid, resources_limits, networking, branch) -> PendingRun

default:"None"

default:"()"

default:"None"

default:"False"

default:"None"

default:"False"

default:"None"

default:"False"

default:"None"

PendingRun

PendingRun

`set_cwd()`

async def set_cwd(cwd) -> None

required

None

`set_env()`

async def set_env(updates) -> None

required

None

`spawn()`

async def spawn(command, *, shell, args, env, cwd, stdin, stdin_open, files, timeout, disposable, truncate_output_at, preserve_env, hostname, uid, gid, resources_limits, networking) -> AsyncOperationContract

default:"None"

default:"()"

default:"None"

default:"False"

default:"None"

default:"False"

default:"None"

default:"False"

default:"None"

AsyncOperationContract

AsyncOperationContract

`spawn_detached()`

async def spawn_detached(command, **kwargs) -> OperationRecord

Spawn and register a command; options are RunRequest fields.

default:"None"

OperationRecord

OperationRecordA durable record whose UUID can be passed to wait_operation.

`spawn_request()`

async def spawn_request(request) -> AsyncOperationContract

Prepare and submit a command, then construct its operation through the factory.

required

AsyncOperationContract

AsyncOperationContractA bound operation whose context records the effective request.

`stage_files()`

async def stage_files(files) -> HistoryEntry

required

HistoryEntry

HistoryEntry

`submit_request()`

async def submit_request(request, image_uuid, *, files, stdin) -> InstanceSpawnResponse

Submit prepared input to the transport. Override for transport-level customization.

required

InstanceSpawnResponse

The transport response identifying the spawned operation.

`switch_branch()`

async def switch_branch(name) -> None

required

None

`wait_operation()`

async def wait_operation(operation_uuid, *, timeout, branch) -> InstanceResult

Wait and finish once, without switching the session’s active branch.Completed outcomes are replayed locally. Transport errors remain pending. Late results retain their original parent without replacing newer local positions. The first completed outcome wins across concurrent policies.

required

default:"None"

InstanceResult

The stored process result, including nonzero exit codes.

`write_file()`

async def write_file(path, content) -> None

required

None

`class ContreeSession`

class contree_sdk.session.ContreeSession(client, *, image, session_id, node_id, tree_kind, store, file_transfer, cwd, env, commit_policy)

A durable, resumable ConTree session backed by a Store (sync).

Attributes

str

Methods

`check_operation_endpoint()`

def check_operation_endpoint(record) -> None

Check the recorded server before using an operation UUID.

required

None

ValueError – The current client targets a different endpoint.

`commit_result()`

def commit_result(operation, *, title, branch, files, data) -> HistoryEntry | None

required

default:"None"

HistoryEntry | None

HistoryEntry | None

`create_branch()`

def create_branch(name, *, from_branch) -> None

required

default:"None"

None

`create_file_transfer()`

def create_file_transfer() -> SyncFileTransfer

Create file I/O when no component was supplied.

SyncFileTransfer

SyncFileTransferA file-transfer component using the session client.

`create_inspector()`

def create_inspector(image_uuid) -> ImageInspector

Create a view of one image. Override to customize inspection.

required

ImageInspector

An inspector borrowing the client’s connection and file transfer.

`create_operation()`

def create_operation(response, context) -> OperationContract

Construct a lifecycle implementation bound to the supplied operation context.

required

OperationContract

OperationContractAn implementation of the matching operation lifecycle contract.

`create_operation_record()`

def create_operation_record(operation) -> OperationRecord

Serialize an operation context. Override to persist custom request fields.

required

OperationRecord

OperationRecordA pending record without local input sources.

ValueError – The operation does not belong to this session.

`create_store()`

def create_store() -> SyncStore

Create the default store when none was supplied.

SyncStore

SyncStoreA new memory store owned by this session.

`delete_branch()`

def delete_branch(name) -> None

required

None

`detach()`

def detach(operation) -> OperationRecord

required

OperationRecord

OperationRecordIts durable record. The caller still owns the live handle.

ValueError – The operation does not belong to this session.

`execute()`

def execute(request, *, branch) -> InstanceResult

Run a request to completion and commit its effective non-disposable result.

required

default:"None"

InstanceResult

The completed process result, including nonzero exit codes.

ValueError – The operation factory returned an unbound operation.

`get_entry()`

def get_entry(node_id) -> HistoryEntry

required

HistoryEntry

HistoryEntry

`get_operation()`

def get_operation(operation_uuid) -> OperationRecord

required

OperationRecord

OperationRecord

`grep()`

def grep(pattern, *, path, glob, max_count, max_total, case, before, after) -> GrepResult

Search session files using the current source’s syntax and limits.

required

default:"None"

GrepResult

Matching lines and the server’s truncation indicator.

`history()`

def history() -> tuple[list[HistoryEntry], dict[str, list[str]]]

tuple[list[HistoryEntry], dict[str, list[str]]]

tuple[list[HistoryEntry], dict[str, list[str]]]

`inspect()`

def inspect() -> ImageInspector

Pin the current saved image without starting a VM.Pending staged attachments appear only after an operation applies them.

ImageInspector

A read-only view that borrows this session’s client and file transfer.

`list_branches()`

def list_branches() -> list[tuple[str, bool]]

list[tuple[str, bool]]

`list_dir()`

def list_dir(path) -> DirectoryList

List session directory entries relative to the filesystem root.

default:"'/'"

DirectoryList

Directory entries and their filesystem metadata.

`list_operations()`

def list_operations(*, pending_only) -> tuple[OperationRecord, ...]

default:"True"

tuple[OperationRecord, ...]

tuple[OperationRecord, …]

`navigate()`

def navigate(node_id) -> None

required

None

`navigate_forward()`

def navigate_forward(steps) -> None

default:"1"

None

`pending_files()`

def pending_files(node_id) -> tuple[StagedFile, ...]

default:"None"

tuple[StagedFile, ...]

tuple[StagedFile, …]

`prepare_environment()`

def prepare_environment(env) -> dict[str, str] | None

Overlay command variables on session defaults without mutating either input.

required

dict[str, str] | None

A fresh mapping, or None when both inputs omit environment variables.

`prepare_request()`

def prepare_request(request) -> RunRequest

Apply defaults before uploads or execution. Override to validate or transform commands.

required

RunRequest

RunRequestThe effective command with session defaults applied.

`read_file()`

def read_file(path) -> bytes

required

bytes

`refresh_from_entry()`

def refresh_from_entry(entry) -> None

required

None

`resolve_image()`

def resolve_image(image) -> str

Resolve a base image. Override to implement an import or registry policy.

required

str

The resolved image UUID.

`restore_operation()`

def restore_operation(operation_uuid) -> OperationContract

Recreate a control handle from the persisted source context.

required

OperationContract

OperationContractAn unentered handle. Use wait_operation to reuse a stored final result.

`restore_operation_context()`

def restore_operation_context(record) -> OperationContext

Decode a stored request. Override together with create_operation_record.

required

OperationContext

OperationContextIts immutable source context, ready for an operation factory or policy.

`rollback()`

def rollback(steps) -> None

default:"1"

None

`run()`

def run(command, *, shell, args, env, cwd, stdin, stdin_open, files, timeout, disposable, truncate_output_at, preserve_env, hostname, uid, gid, resources_limits, networking, branch) -> InstanceResult

default:"None"

default:"()"

default:"None"

default:"False"

default:"None"

default:"False"

default:"None"

default:"False"

default:"None"

InstanceResult

`set_cwd()`

def set_cwd(cwd) -> None

required

None

`set_env()`

def set_env(updates) -> None

required

None

`spawn()`

def spawn(command, *, shell, args, env, cwd, stdin, stdin_open, files, timeout, disposable, truncate_output_at, preserve_env, hostname, uid, gid, resources_limits, networking) -> OperationContract

default:"None"

default:"()"

default:"None"

default:"False"

default:"None"

default:"False"

default:"None"

default:"False"

default:"None"

OperationContract

OperationContract

`spawn_detached()`

def spawn_detached(command, **kwargs) -> OperationRecord

Spawn and register a command; options are RunRequest fields.

default:"None"

OperationRecord

OperationRecordA durable record whose UUID can be passed to wait_operation.

`spawn_request()`

def spawn_request(request) -> OperationContract

Prepare and submit a command, then construct its operation through the factory.

required

OperationContract

OperationContractA bound operation whose context records the effective request.

`stage_files()`

def stage_files(files) -> HistoryEntry

required

HistoryEntry

HistoryEntry

`submit_request()`

def submit_request(request, image_uuid, *, files, stdin) -> InstanceSpawnResponse

Submit prepared input to the transport. Override for transport-level customization.

required

InstanceSpawnResponse

The transport response identifying the spawned operation.

`switch_branch()`

def switch_branch(name) -> None

required

None

`wait_operation()`

def wait_operation(operation_uuid, *, timeout, branch) -> InstanceResult

required

default:"None"

InstanceResult

The stored process result, including nonzero exit codes.

`write_file()`

def write_file(path, content) -> None

Write bytes to an absolute path, creating its parent directories.Returns after the write is applied to the executor’s filesystem. It does not merely stage a file or upload a blob. Persistence follows execute(). The default runs mkdir and cat through execute() with binary stdin. Override this method to use another file-transfer mechanism. An override must coordinate with the executor’s commands and snapshot lifecycle.Invalid paths raise ValueError; a failed write raises FileWriteError. Transport errors, timeouts, and cancellation propagate unchanged.

required

None

`class DurationSnapshotPolicy`

class contree_sdk.session.DurationSnapshotPolicy(seconds, *, timer_factory, clock)

Request a snapshot after a fixed VM lifetime, including startup time.

Methods

`expired()`

def expired(generation) -> None

required

None

`notify()`

def notify(event) -> None

Handle an execution event without blocking or performing network I/O.

required

None

`schedule_timer()`

def schedule_timer() -> None

None

`set_duration()`

def set_duration(seconds) -> None

Update the deadline relative to the original VM startup time.

required

None

ValueError – The duration is not finite and positive.

`should_snapshot()`

def should_snapshot() -> Future[None]

Future[None]

A future shared by all readers of the current policy decision.

`start_deferred()`

def start_deferred() -> None

Record startup time; set_duration later arms the effective deadline.

None

`class HistoryEntry`

class contree_sdk.session.HistoryEntry(id, session_id, image_uuid, parent_id, kind, title, operation_uuid, exit_code, created_at, files, attachments, applied_files, data)

Attributes

tuple[str, ...]

tuple[StagedFile, ...]

tuple[StagedFile, …]

tuple[str, ...]

str

str | None

str

str | None

int | None

datetime

`class IdleSnapshotPolicy`

class contree_sdk.session.IdleSnapshotPolicy(seconds, *, timer_factory)

Request a snapshot after commands and inspections have been idle for seconds. Timers start when the last command or inspection finishes. Admission, reset, or close cancels the timer. Async callers use their event loop’s timer; sync callers use a daemon threading.Timer. Inject timer_factory for a custom scheduler.

Methods

`notify()`

def notify(event) -> None

Handle an execution event without blocking or performing network I/O.

required

None

`should_snapshot()`

def should_snapshot() -> Future[None]

Future[None]

A future shared by all readers of the current policy decision.

`class LazySession`

class contree_sdk.session.LazySession(session, *, snapshot_policy, snapshot_timeout, vm_timeout)

Attributes

BaseException | None

Literal['cold', 'starting', 'running', 'draining', 'snapshotting', 'failed', 'closed']

str

Methods

`abort()`

def abort() -> None

Discard unsaved work and close. An already running snapshot finishes first.

None

`close()`

def close(*, timeout) -> None

Save pending work and close this wrapper, retaining caller-owned resources.

default:"None"

None

RuntimeError – Saving failed; the cause is available as error.

TimeoutError – The caller’s wait expired; closing continues in the worker.

`create_branch()`

def create_branch(name, *, from_branch) -> None

required

default:"None"

None

`execute()`

def execute(request) -> InstanceResult

required

InstanceResult

`grep()`

def grep(pattern, *, path, glob, max_count, max_total, case, before, after) -> GrepResult

Search session files using the current source’s syntax and limits.

required

default:"None"

GrepResult

Matching lines and the server’s truncation indicator.

`inspect()`

def inspect(*, timeout) -> SessionInspector

default:"None"

SessionInspector

A session-bound inspector; creating it starts no VM.

`inspect_snapshot()`

def inspect_snapshot(*, timeout) -> ImageInspector

default:"None"

ImageInspector

An inspector borrowing the underlying client and file transfer.

`inspection_source()`

def inspection_source(*, timeout) -> Iterator[ImageInspector | str]

default:"None"

Iterator[ImageInspector | str]

A saved image inspector or an active operation UUID.

RuntimeError – The session fails or closes before inspection starts.

TimeoutError – Admission or VM startup exceeds the timeout.

`keepalive_request()`

def keepalive_request() -> RunRequest

Supply a long-running main process available in the base image.

RunRequest

RunRequestThe non-disposable request used to start each VM.

`list_dir()`

def list_dir(path) -> DirectoryList

List session directory entries relative to the filesystem root.

default:"'/'"

DirectoryList

Directory entries and their filesystem metadata.

`observe()`

def observe(event, error) -> None

Observe the operation’s sole reader without depending on handle.wait().

required

None

`prepare_request()`

def prepare_request(request) -> RunRequest

Apply the wrapped session’s policy and validate shared-VM semantics.

required

RunRequest

RunRequestThe effective request.

ValueError – The request needs a disposable VM or startup-only options.

`read_file()`

def read_file(path) -> bytes

Read current files without stopping an active VM.

required

bytes

The complete file as bytes.

`rollback()`

def rollback(steps) -> None

default:"1"

None

`run()`

def run(command, **kwargs) -> InstanceResult

Run a shared-VM command. Options are RunRequest fields; disposable defaults to False.

default:"None"

InstanceResult

The subprocess result, including a nonzero exit code.

`snapshot()`

def snapshot(*, timeout) -> HistoryEntry | None

default:"None"

HistoryEntry | None

HistoryEntry | NoneThe saved history entry, or None if no VM has run yet.

`spawn()`

def spawn(command, **kwargs) -> SubprocessContract

default:"None"

SubprocessContract

SubprocessContract

`spawn_request()`

def spawn_request(request) -> SubprocessContract

Start a subprocess; track its exit even if the caller never waits.

required

SubprocessContract

SubprocessContractThe subprocess handle from the operation factory.

`switch_branch()`

def switch_branch(name) -> None

required

None

`write_file()`

def write_file(path, content) -> None

required

None

`class Operation`

class contree_sdk.session.Operation(client, operation_uuid, *, timeout, shutdown_timeout, files, context)

Handle to a spawned instance operation, identified by its UUID.

Attributes

str

OperationContext | None

OperationContext | None

OperationResponse | None

Methods

`add_observer()`

def add_observer(observer) -> None

required

None

`apply_files()`

def apply_files(files) -> None

Apply already-uploaded files to this running VM.

required

None

`cancel()`

def cancel() -> None

None

`claim_queue()`

def claim_queue(spid) -> Queue[OperationEvent | None]

required

Queue[OperationEvent | None]

`create_subprocess()`

def create_subprocess(spid, events_queue) -> SubprocessContract

Create a subprocess handle. Override to return a custom implementation.

required

SubprocessContract

SubprocessContractA handle consuming the supplied event queue.

`events()`

def events(*, since, spid, timeout) -> Iterator[OperationEvent]

Replay buffered events, then follow the shared reader with local filters.timeout bounds this subscription. Expiry does not cancel the operation.

default:"None"

Iterator[OperationEvent]

Events after since whose process matches spid, when specified.

TimeoutError – The subscription did not finish before timeout.

`open_event_stream()`

def open_event_stream() -> Iterator[OperationEvent]

Open the unfiltered source. Override this hook to supply custom events.

Iterator[OperationEvent]

The single transport stream consumed by the operation.

`pipe_stdin()`

def pipe_stdin(chunks, *, spid, close, chunk_size) -> StdinResult

Forward chunks after spawn, using the operation’s shared event reader.The caller owns the source and must make blocking reads interruptible. Use one writer per process. Errors stop the target process and propagate. Errors for spid=1 cancel the operation.

required

default:"1"

default:"True"

default:"65536"

StdinResult

StdinResultAcknowledged bytes, EOF delivery, and observed early completion.

ValueError – chunk_size or spid is not positive.

`pump()`

def pump() -> None

None

`route_event()`

def route_event(event) -> None

required

None

`run()`

def run(command, *, shell, args, env, cwd, uid, gid, stdin, stdin_open, truncate_output_at) -> SubprocessContract

required

default:"False"

default:"()"

default:"None"

default:"False"

default:"None"

SubprocessContract

SubprocessContract

`send_stdin()`

def send_stdin(data, *, spid, close) -> None

required

default:"1"

default:"True"

None

`shutdown()`

def shutdown() -> None

None

`signal()`

def signal(sig, *, spid) -> None

default:"None"

default:"1"

None

`status()`

def status(*, inflight) -> OperationResponse

default:"False"

OperationResponse

`wait()`

def wait(*, timeout) -> InstanceResult

default:"None"

InstanceResult

`class OperationContext`

class contree_sdk.session.OperationContext(request, session_id, image_uuid, parent_id, branch, files, attachments, detached, node_id, created_at, position_version, session_cwd, session_env)

The effective request and history position captured before remote execution.

Attributes

tuple[StagedFile, ...]

tuple[StagedFile, …]

bool

tuple[str, ...]

int

str | None

RunRequest

RunRequest

str

str | None

str

float

Mapping[str, str]

`class OperationContract`

class contree_sdk.session.OperationContract()

An operation lifecycle. wait stores response before returning; cleanup preserves primary errors.

Attributes

str

OperationContext | None

OperationContext | None

OperationResponse | None

Methods

`add_observer()`

def add_observer(observer) -> None

required

None

`apply_files()`

@abstractmethod
def apply_files(files) -> None

required

None

`cancel()`

@abstractmethod
def cancel() -> None

None

`events()`

@abstractmethod
def events(*, since, spid, timeout) -> Iterator[OperationEvent]

default:"None"

Iterator[OperationEvent]

`pipe_stdin()`

@abstractmethod
def pipe_stdin(chunks, *, spid, close, chunk_size) -> StdinResult

required

default:"1"

default:"True"

default:"65536"

StdinResult

StdinResult

`run()`

@abstractmethod
def run(command, *, shell, args, env, cwd, uid, gid, stdin, stdin_open, truncate_output_at) -> SubprocessContract

required

default:"False"

default:"()"

default:"None"

default:"False"

default:"None"

SubprocessContract

SubprocessContract

`send_stdin()`

@abstractmethod
def send_stdin(data, *, spid, close) -> None

required

default:"1"

default:"True"

None

`shutdown()`

@abstractmethod
def shutdown() -> None

None

`signal()`

@abstractmethod
def signal(sig, *, spid) -> None

default:"None"

default:"1"

None

`status()`

@abstractmethod
def status(*, inflight) -> OperationResponse

default:"False"

OperationResponse

`wait()`

@abstractmethod
def wait(*, timeout) -> InstanceResult

default:"None"

InstanceResult

`class OperationRecord`

class contree_sdk.session.OperationRecord(*, uuid, session_id, image_uuid, parent_id, branch, title, disposable, request, endpoint, created_at, node_id, session_cwd, session_env, files, attachments, response, history_id)

A registered operation and its first atomically completed outcome. request holds the effective command, excluding source streams. A non-null response marks completion, including failed, cancelled, and discarded results. history_id is set only when that completion retained an image.

Attributes

tuple[StagedFile, ...]

tuple[StagedFile, …]

float

str

tuple[str, ...]

str | None

str

bool

str | None

str

bool

dict[str, str]

Methods

`from_dict()`

@classmethod
def from_dict(payload) -> OperationRecord

required

OperationRecord

OperationRecord

`to_dict()`

def to_dict() -> dict[str, str | int | float | bool | None | list[str | int | float | bool | None | list[JSONValue] | dict[str, JSONValue]] | dict[str, str | int | float | bool | None | list[JSONValue] | dict[str, JSONValue]]]

`class PendingRun`

class contree_sdk.session.PendingRun(session, request, *, branch)

One-use execution wrapper. Await a result or enter an operation context.

Methods

`commit()`

async def commit(operation) -> None

required

None

`run_to_result()`

async def run_to_result() -> InstanceResult

InstanceResult

`spawn()`

async def spawn() -> AsyncOperationContract

AsyncOperationContract

AsyncOperationContract

`class RunRequest`

class contree_sdk.session.RunRequest(*, command, shell, args, env, cwd, stdin, stdin_open, files, timeout, disposable, truncate_output_at, preserve_env, hostname, uid, gid, resources_limits, networking)

Command configuration passed through the session’s public extension hooks. Use dataclasses.replace to change a request. Arguments and environment are copied at construction. File sources and stdin remain caller-owned. Exactly one of command and shell must be supplied. Results are saved by default; use disposable=True to leave session history unchanged.

Attributes

tuple[str, ...]

str | None

bool

Mapping[str, str] | None

int | None

str | None

InstanceNetworking | None

bool

InstanceResourcesLimits | None

str | None

bool

float | timedelta | None

float | None

str

int | None

`class SnapshotEvent`

class contree_sdk.session.SnapshotEvent(*values)

Notifications for one policy instance owned by one lazy session. RESET starts a new VM cycle. VM_STARTING precedes the VM spawn request. COMMAND_STARTED reserves an accepted command before its subprocess is created. COMMAND_FINISHED confirms its exit event, including nonzero exits. INSPECTION_STARTED/FINISHED bracket a live filesystem request, including its stream. They affect idle time, not command counts. CLOSE releases resources after completion or failure.

Attributes

`class StdinResult`

class contree_sdk.session.StdinResult(bytes_sent, eof_sent, process_exited)

Acknowledged input bytes and whether EOF was sent or the process ended. A successful write does not prove that the remote process read the bytes. An ambiguous transport error raises instead of returning a partial result.

Attributes

int

bool

`class SubprocessContract`

class contree_sdk.session.SubprocessContract()

Waitable event stream for one subprocess. Iteration and wait share completion state.

Attributes

int

Methods

`pipe_to()`

@abstractmethod
def pipe_to(*, stdout, stderr) -> InstanceResult

default:"None"

InstanceResult

`wait()`

@abstractmethod
def wait(*, timeout) -> InstanceResult

default:"None"

InstanceResult

`class SubprocessHandle`

class contree_sdk.session.SubprocessHandle(operation, spid, events_queue)

Both blocking-waitable (final result) and iterable (live events) for one spid.

Attributes

int

Methods

`pipe_to()`

def pipe_to(*, stdout, stderr) -> InstanceResult

default:"None"

InstanceResult

`wait()`

def wait(*, timeout) -> InstanceResult

default:"None"

InstanceResult

`class ZeroExitCommitPolicy`

class contree_sdk.session.ZeroExitCommitPolicy()

Retain a result only when its process exit code is explicitly zero.

Methods

`should_commit()`

def should_commit(context, result) -> bool

required

bool

`instance_result()`

def contree_sdk.session.instance_result(op) -> InstanceResult

Get the operation’s InstanceResult.

required

InstanceResult

The InstanceResult from op.metadata.result.

InterruptedError – The operation was cancelled.

FailedOperationError – The operation itself failed with no result at all (a nonzero exit code inside a successful operation is not an error).

​class AbstractCommitPolicy

​Methods

​should_commit()

​class AbstractSnapshotPolicy

​Methods

​notify()

​should_snapshot()

​class ApiSuccessCommitPolicy

​Methods

​should_commit()

​class AsyncLazySession

​Attributes

​Methods

​abort()

​close()

​create_branch()

​execute()

​grep()

​inspect()

​inspect_snapshot()

​inspection_source()

​keepalive_request()

​list_dir()

​observe()

​prepare_request()

​read_file()

​rollback()

​run()

​snapshot()

​spawn()

​spawn_request()

​switch_branch()

​write_file()

​class AsyncOperation

​Attributes

​Methods

​add_observer()

​apply_files()

​cancel()

​claim_queue()

​create_subprocess()

​events()

​open_event_stream()

​pipe_stdin()

​pump()

​route_event()

​run()

​send_stdin()

​shutdown()

​signal()

​status()

​wait()

​class AsyncOperationContract

​Attributes

​Methods

​add_observer()

​apply_files()

​cancel()

​events()

​pipe_stdin()

​run()

​send_stdin()

​shutdown()

​signal()

​status()

​wait()

​class AsyncSubprocessContract

​Attributes

​Methods

​pipe_to()

​wait()

​class AsyncSubprocessHandle

​Attributes

​Methods

​iterate()

​pipe_to()

​wait()

​class CommandCountSnapshotPolicy

​Methods

​notify()

`class AbstractCommitPolicy`

Methods

`should_commit()`

`class AbstractSnapshotPolicy`

Methods

`notify()`

`should_snapshot()`

`class ApiSuccessCommitPolicy`

Methods

`should_commit()`

`class AsyncLazySession`

Attributes

Methods

`abort()`

`close()`

`create_branch()`

`execute()`

`grep()`

`inspect()`

`inspect_snapshot()`

`inspection_source()`

`keepalive_request()`

`list_dir()`

`observe()`

`prepare_request()`

`read_file()`

`rollback()`

`run()`

`snapshot()`

`spawn()`

`spawn_request()`

`switch_branch()`

`write_file()`

`class AsyncOperation`

Attributes

Methods

`add_observer()`

`apply_files()`

`cancel()`

`claim_queue()`

`create_subprocess()`

`events()`

`open_event_stream()`

`pipe_stdin()`

`pump()`

`route_event()`

`run()`

`send_stdin()`

`shutdown()`

`signal()`

`status()`

`wait()`

`class AsyncOperationContract`

Attributes

Methods

`add_observer()`

`apply_files()`

`cancel()`

`events()`

`pipe_stdin()`

`run()`

`send_stdin()`

`shutdown()`

`signal()`

`status()`

`wait()`

`class AsyncSubprocessContract`

Attributes

Methods

`pipe_to()`

`wait()`

`class AsyncSubprocessHandle`

Attributes

Methods

`iterate()`

`pipe_to()`

`wait()`

`class CommandCountSnapshotPolicy`

Methods

`notify()`