---
icon: diagram-project
---

# Advanced: control a running sandbox

For ordinary commands, start with {doc}`getting-started` and its sandbox factory.
This guide exposes lower-level controls. See {doc}`manual-sandbox` for manual
component assembly and resource ownership.

Use `run()` when you only need a completed result. Use `spawn()` to obtain an
operation handle immediately, stream output, send input, or cancel work.
Enter an operation context to start additional processes in the same running sandbox.
Use {doc}`detached-operations` to save its context and collect the result after
restarting Python.

:::{note}
Async snippets with top-level `await` run inside an async function or a notebook
that supports it. For a standalone script, use the `asyncio.run(main())` structure
from {doc}`getting-started`.
:::

## Run an additional process

This program keeps a sandbox alive with `sleep`, runs `echo` inside it, and streams
the output locally. Configure a connection as described in {ref}`sandbox-connection`.

::::{tab} Sync

<!--
name: test_processes; fixtures: doc_api, capsys
```python
doc_api.subprocess()
```
-->

```python
"""Run a subprocess inside an operation."""

import sys

from contree_sdk import ContreeSession, create_sandbox_session


def main() -> None:
    with create_sandbox_session(session_type=ContreeSession, image="tag:alpine:latest") as session:
        operation = session.spawn(shell="sleep 300", timeout=60)
        with operation:
            process = operation.run("echo", args=["Hello from a subprocess!"])
            process.pipe_to(stdout=sys.stdout, stderr=sys.stderr)


main()
```

<!--
name: test_processes
```python
assert doc_api.sync.calls_for("operation_subprocess_create")[0].args == ("operation-1", "echo")
assert doc_api.sync.calls_for("operation_subprocess_create")[0].kwargs["args"] == ["Hello from a subprocess!"]
assert capsys.readouterr().out == "Hello from a subprocess!\n"
```
-->

::::

::::{tab} Async

<!--
name: test_processes_async; fixtures: doc_api, capsys
```python
doc_api.subprocess()
```
-->

```python
import asyncio
import sys

from contree_sdk import ContreeAsyncSession, create_sandbox_session


async def main() -> None:
    async with create_sandbox_session(session_type=ContreeAsyncSession, image="tag:alpine:latest") as session:
        operation = await session.spawn(shell="sleep 300", timeout=60)
        async with operation:
            process = await operation.run("echo", args=["Hello from a subprocess!"])
            await process.pipe_to(stdout=sys.stdout, stderr=sys.stderr)


asyncio.run(main())
```

<!--
name: test_processes_async
```python
assert doc_api.async_client.calls_for("operation_subprocess_create")[0].args == ("operation-1", "echo")
assert doc_api.async_client.calls_for("operation_subprocess_create")[0].kwargs["args"] == ["Hello from a subprocess!"]
assert capsys.readouterr().out == "Hello from a subprocess!\n"
```
-->

::::

On context exit, the SDK signals the main process, waits for shutdown, and attempts
cancellation if it does not stop. This happens even when the block raises an exception.
The subprocess shares the operation's live filesystem. It has its own `spid` and
result, but no separate session-history entry.

To place already-uploaded files in a running operation, pass the prepared
path-to-`FileSpec` mapping to `operation.apply_files()` (await it for an async
operation). For example, `operation.apply_files(session.file_transfer.prepare_files({"/input": b"data"}) or {})`.
The call replaces existing files by default. Live upload uses the VM runtime user
and mode `0644`; non-default `FileSpec` ownership or mode is rejected. Each file
is applied separately, so a later failure does not undo earlier attachments.

## Choose the right handle

| Call                                           | What you receive                                  |
| ---------------------------------------------- | ------------------------------------------------- |
| `session.run(...)`                             | A completed `InstanceResult`.                     |
| `session.spawn(...)`                           | An `OperationContract` for the main process.      |
| `operation.run(...)` inside a context          | A `SubprocessContract` for an additional process. |
| `await session.run(...)`                       | A completed `InstanceResult` in async code.       |
| `await session.spawn(...)`                     | An `AsyncOperationContract`.                      |
| `await operation.run(...)` inside `async with` | An `AsyncSubprocessContract`.                     |

Async `session.run()` also supports `async with`. Its pending call can be consumed
once. A sync `run()` returns a result and cannot be used as an operation context.

## Stream the main process

Iterate `operation.events()` when you need live output without additional processes.
Print stream payloads directly; `print()` adds a newline after each payload.

::::{tab} Sync

<!--
name: test_stream_events; fixtures: doc_api, capsys
```python
doc_api.subprocess(stdout="hello\n")
```
-->

```python
from contree_client.models import EventDataStream

from contree_sdk import ContreeSession, create_sandbox_session


with create_sandbox_session(session_type=ContreeSession, image="tag:alpine:latest") as session:
    operation = session.spawn(shell="echo hello")
    for event in operation.events():
        if event.type in {"stdout", "stderr"} and isinstance(event.data, EventDataStream):
            print(event.data)
    operation.wait()
```

<!--
name: test_stream_events
```python
assert capsys.readouterr().out == "hello\n\n"
assert doc_api.sync.calls_for("follow_operation_events")[0].args == ("operation-1",)
assert len(doc_api.sync.calls_for("follow_operation_events")) == 1
assert not doc_api.sync.calls_for("wait_operation")
```
-->

::::

::::{tab} Async

<!--
name: async test_stream_events_async; fixtures: doc_api, capsys
```python
doc_api.subprocess(stdout="hello\n")
```
-->

```python
from contree_client.models import EventDataStream

from contree_sdk import ContreeAsyncSession, create_sandbox_session


async with create_sandbox_session(session_type=ContreeAsyncSession, image="tag:alpine:latest") as session:
    operation = await session.spawn(shell="echo hello")
    async for event in operation.events():
        if event.type in {"stdout", "stderr"} and isinstance(event.data, EventDataStream):
            print(event.data)
    await operation.wait()
```

<!--
name: test_stream_events_async
```python
assert capsys.readouterr().out == "hello\n\n"
assert doc_api.async_client.calls_for("follow_operation_events")[0].args == ("operation-1",)
assert len(doc_api.async_client.calls_for("follow_operation_events")) == 1
assert not doc_api.async_client.calls_for("wait_operation")
```
-->

::::

## One reader per operation handle

`events()`, `wait()`, and subprocess handles share one unfiltered SSE reader.
It starts when the context is entered or when the first event iterator or wait is
consumed. A sync operation uses one reader thread; an async operation uses one task.
Transport reconnection remains the responsibility of `contree-client`.

Each `events()` iterator has an independent cursor. Events are retained in memory
for the lifetime of the handle, so a late subscriber can replay them without another
network stream. `since=N` selects event IDs greater than `N`; `spid=N` selects that
process. A process filter excludes the operation-wide `completion` event, but the
iterator still ends when the shared reader ends. Large event logs consume memory;
release completed handles when replay is no longer needed.

`wait()` uses the same reader and then fetches the final status once. Repeated waits
reuse that response. Closing or cancelling an event iterator only removes that
subscriber. A failed or cancelled operation `wait()` attempts remote cancellation
and therefore affects all consumers of the operation. Call `shutdown()` if you
stop consuming before completion and no longer need the operation.

Session `run()`, `execute()`, and `wait_operation()` and Docker builds close the
operation handles they create. Async cleanup waits for the reader to stop, even
if the caller cancels again during cleanup. Cleanup can extend a timeout while
cancellation requests and reader shutdown finish. These helpers do not close a
client supplied by the caller.

Synchronous shutdown waits for the reader thread for a bounded time. It cannot
interrupt an arbitrary blocking event source. If the source remains blocked,
the thread exits only when that source returns or raises.

Sharing applies to one `Operation` or `AsyncOperation` object. Constructing another
handle for the same UUID creates a separate owner and can open another stream.
Reuse the existing handle when consumers run in the same application.

## Keep the result image

`spawn()` retains the result image by default. Wait for completion, then call
`commit_result(operation)` to advance session history. `run()` commits automatically.
Use `disposable=True` to leave the session image unchanged.

Saving an image is lightweight and can happen after every command. It stops the
VM; if files have not changed, the result can have the same image UUID.
::::{tab} Sync

<!--
name: test_manual_commit; fixtures: doc_api, capsys
```python
doc_api.complete()
```
-->

```python
from contree_sdk import ContreeSession, create_sandbox_session


with create_sandbox_session(session_type=ContreeSession, image="tag:alpine:latest") as session:
    operation = session.spawn(shell="echo saved > /saved.txt")
    result = operation.wait()
    entry = session.commit_result(operation, title="Save output")
    print(entry.image_uuid)
```

<!--
name: test_manual_commit
```python
assert session.image_uuid == "image-1"
assert entry.operation_uuid == "operation-1"
assert entry.title == "Save output"
assert capsys.readouterr().out == "image-1\n"
```
-->

::::

::::{tab} Async

<!--
name: async test_manual_commit_async; fixtures: doc_api, capsys
```python
doc_api.complete()
```
-->

```python
from contree_sdk import ContreeAsyncSession, create_sandbox_session


async with create_sandbox_session(session_type=ContreeAsyncSession, image="tag:alpine:latest") as session:
    operation = await session.spawn(shell="echo saved > /saved.txt")
    result = await operation.wait()
    entry = await session.commit_result(operation, title="Save output")
    print(entry.image_uuid)
```

<!--
name: test_manual_commit_async
```python
assert session.image_uuid == "image-1"
assert entry.operation_uuid == "operation-1"
assert entry.title == "Save output"
assert capsys.readouterr().out == "image-1\n"
```
-->

::::

A subprocess's writes can be part of that operation's final filesystem image; only
the whole operation creates a history entry.

A UUID-only handle can inspect, wait for, or cancel an existing operation. It has no
session history context, so `commit_result()` cannot use it. Keep the original
session-created handle when you intend to commit.

A subprocess remembers its exit event. `wait()` after iteration does not wait for
another exit. Stream failures propagate to consumers. Operation `wait()` attempts
remote cancellation if waiting fails or is cancelled, while preserving the original error.

## Stream input with bounded memory

Start the process with `stdin_open=True`. This keeps remote stdin open until you
send EOF. `pipe_stdin()` waits for that process's `spawn` event, sends chunks in
order, and sends an empty EOF request when the source ends. Its default maximum
input chunk size is 64 KiB after UTF-8 encoding. Base64 transport adds overhead.

Supply an iterable of `str` or `bytes` in sync code, or an async iterable in async
code. The SDK holds one source chunk at a time and splits large chunks before
sending them. To bound total input memory, keep the source's chunks bounded too.
`stdin=...` remains an eager initial payload; use `pipe_stdin()` for large files.

::::{tab} Sync

<!--
name: test_operation_input; fixtures: stdin_api
-->

```python
from functools import partial
from pathlib import Path

from contree_sdk import ContreeSession, create_sandbox_session


# Create sample input; replace it with your own file when using this workflow.
Path("input.bin").write_bytes(bytes(range(256)) * 1024)

with create_sandbox_session(session_type=ContreeSession, image="tag:alpine:latest") as session:
    operation = session.spawn("cat", stdin_open=True)
    with open("input.bin", "rb") as source:
        delivery = operation.pipe_stdin(iter(partial(source.read, 65536), b""))
    result = operation.wait()
```

<!--
name: test_operation_input
```python
from pathlib import Path

assert delivery.eof_sent
assert delivery.bytes_sent == Path("input.bin").stat().st_size
assert bytes(stdin_api.sync.received) == Path("input.bin").read_bytes()
assert len(stdin_api.sync.calls_for("follow_operation_events")) == 1
assert stdin_api.sync.calls_for("spawn_instance")[0].kwargs["stdin"].close is False
```
-->

::::

::::{tab} Async

<!--
name: async test_operation_input_async; fixtures: stdin_api
-->

```python
import asyncio
from pathlib import Path

from contree_sdk import ContreeAsyncSession, create_sandbox_session


# Create sample input before starting the remote process.
Path("input.bin").write_bytes(bytes(range(256)) * 1024)


async def input_chunks():
    source = await asyncio.to_thread(open, "input.bin", "rb")
    try:
        while chunk := await asyncio.to_thread(source.read, 65536):
            yield chunk
    finally:
        await asyncio.to_thread(source.close)


async with create_sandbox_session(session_type=ContreeAsyncSession, image="tag:alpine:latest") as session:
    operation = await session.spawn("cat", stdin_open=True)
    source = input_chunks()
    try:
        delivery = await operation.pipe_stdin(source)
    finally:
        await source.aclose()
    result = await operation.wait()
```

<!--
name: test_operation_input_async
```python
from pathlib import Path

assert delivery.eof_sent
assert delivery.bytes_sent == Path("input.bin").stat().st_size
assert bytes(stdin_api.async_client.received) == Path("input.bin").read_bytes()
assert len(stdin_api.async_client.calls_for("follow_operation_events")) == 1
assert stdin_api.async_client.calls_for("spawn_instance")[0].kwargs["stdin"].close is False
```
-->

::::

`StdinResult.bytes_sent` counts bytes acknowledged by the server. It does not
prove that the process consumed them. `eof_sent` records successful EOF delivery.
If the shared reader observes an exit or operation completion before delivery
finishes, `process_exited` is true and the SDK stops sending input.

The caller owns the input source. A sync iterator can block inside `next()`;
the SDK checks for exit before and after that read but cannot interrupt arbitrary
blocking Python code. Use an interruptible source for live TTY input. Async
forwarding cancels a pending source read when the process exits. Sources must
cooperate with cancellation; cancelling `asyncio.to_thread()` does not stop its
underlying thread. The file example uses finite reads from a regular file.

Input-source errors, transport errors, and cancellation cancel the operation and
preserve the original exception. The SDK never retries stdin writes: a timeout
can mean that some bytes were delivered. An exit can race with a write and cause
a transport error instead of a `process_exited` result.

Use one writer per process. Set `close=False` on `pipe_stdin()` to deliver several
input batches before EOF. For a child process, use
`operation.run(..., stdin_open=True)` and pass its `spid` to `pipe_stdin()`.
`send_stdin()` remains a low-level single-write method; it does not wait for spawn
or manage input-source errors. Its default `close=True` sends EOF after that write.

`pipe_stdin()` returns after input delivery, without waiting for the operation.
Save `operation.uuid` if another owner will wait for completion. Do not enter an
operation context for this handoff: context exit stops the main process. The
shared reader remains active until completion or client closure. Input forwarding,
`events()`, and `wait()` on one handle use one transport subscription.
An optional `events(timeout=...)` limits that subscription. Timeout stops only
that subscriber; call `cancel()` to stop the remote operation. The client
resumes that subscription after a connection loss; input is not replayed.

`signal()` sends a signal without waiting. `cancel()` requests operation
cancellation. `wait()` returns the typed `InstanceResult`, including nonzero
process exit codes. A cancelled operation raises `InterruptedError`; an operation
failure raises `FailedOperationError`.

## Reattach to an operation by UUID

Save `operation.uuid` when another process must inspect or cancel work.
Construct a handle with the same UUID and a client for the same server. This does
not start a new operation or restore session history context.

::::{tab} Sync

<!--
name: test_reattach; fixtures: doc_api, capsys
```python
doc_api.complete()
```
-->

```python
from contree_client import profiles
from contree_client.sync import ContreeClient

from contree_sdk import ContreeSession
from contree_sdk.session import Operation


with ContreeClient.from_profile(profiles.from_env() or profiles.resolve_profile()) as client:
    session = ContreeSession(client, image="tag:alpine:latest")
    operation = session.spawn("echo", args=["hello"])
    saved_uuid = operation.uuid
    reattached = Operation(client, saved_uuid)
    result = reattached.wait()
    print(reattached.uuid)
```

<!--
name: test_reattach
```python
assert len(doc_api.sync.calls_for("spawn_instance")) == 1
assert reattached.context is None
assert capsys.readouterr().out == "operation-1\n"
```
-->

::::

::::{tab} Async

<!--
name: async test_reattach_async; fixtures: doc_api, capsys
```python
doc_api.complete()
```
-->

```python
from contree_client import profiles
from contree_client.asyncio import ContreeAsyncClient

from contree_sdk import ContreeAsyncSession
from contree_sdk.session import AsyncOperation


async with ContreeAsyncClient.from_profile(profiles.from_env() or profiles.resolve_profile()) as client:
    session = ContreeAsyncSession(client, image="tag:alpine:latest")
    operation = await session.spawn("echo", args=["hello"])
    saved_uuid = operation.uuid
    reattached = AsyncOperation(client, saved_uuid)
    result = await reattached.wait()
    print(reattached.uuid)
```

<!--
name: test_reattach_async
```python
assert len(doc_api.async_client.calls_for("spawn_instance")) == 1
assert reattached.context is None
assert capsys.readouterr().out == "operation-1\n"
```
-->

::::