cayu.workspaces.base

Public Cayu API. Definitions are grouped by their public source module, and definitions from private implementation modules appear on their package’s page; adding a symbol does not move existing links. The import shown at each entry is a supported public namespace. Read the Python API overview and install the required extra.

`BoundedTarReader`

from cayu.workspaces import BoundedTarReader

Nominal capability for preflight-bounded bulk tar reads.

Implementations must validate the requested files’ combined logical size and the conservative raw archive size before allocating or materializing the archive. Merely accepting the limit keywords or rejecting after construction does not satisfy this contract.

Bases: ABC.

read_tar_bytes (method)

async BoundedTarReader.read_tar_bytes(paths: Sequence[str], *, max_file_bytes: int | None=None, max_total_bytes: int | None=None, max_archive_bytes: int | None=None) -> bytes

Return an uncompressed tar after preflighting every configured limit.

Release source · Runtime guarantees · Capability guides

`BoundedTarStreamReader`

from cayu.workspaces import BoundedTarStreamReader

Nominal capability for bounded tar production into a caller-owned stream.

The destination is a binary stream, never a pathname. Implementations must preflight logical and archive limits before emitting archive bytes and must settle delegated reads before returning.

Bases: ABC.

read_tar_stream (method)

async BoundedTarStreamReader.read_tar_stream(paths: Sequence[str], destination: BinaryIO, *, max_file_bytes: int | None=None, max_total_bytes: int | None=None, max_archive_bytes: int | None=None) -> TarStreamReadResult

Write one uncompressed tar to destination and report its size.

Release source · Runtime guarantees · Capability guides

`matches_list_pattern`

from cayu.workspaces import matches_list_pattern

Report whether a workspace-relative POSIX file path matches a list pattern.

matches_list_pattern(path: str, pattern: str) -> bool

Release source · Runtime guarantees · Capability guides

`RunnerBoundWorkspace`

from cayu.workspaces import RunnerBoundWorkspace

Workspace whose operations target one declared runner-owned resource.

Native bindings use this nominal contract instead of reflecting on an incidental runner attribute. Implementations can prove that they target the lifecycle-owning runner without exposing that runner through the workspace’s public surface.

Bases: Workspace.

Fields and attributes

Defaults below are source expressions; Field(...) includes schema/validation settings and default_factory creates a fresh value. Inherited declarations are included. Constructor behavior follows the model or dataclass shown in the bases/decorators.

Name	Type	Default
`id`	`str`	`required`

reference_binding (method)

RunnerBoundWorkspace.reference_binding() -> WorkspaceReferenceBinding

Bind application data to this workspace incarnation.

The default is object-lifetime scoped and fails closed after reopening. Durable adapters may override this method only when they can preserve an incarnation across recovery and rotate it on replacement. Never restore a generation from the application reference being validated.

read_bytes (method)

async RunnerBoundWorkspace.read_bytes(path: str, *, offset: int=0, max_bytes: int | None=None) -> WorkspaceReadResult

Read a byte page from the workspace.

Complete offset-zero snapshots expose an opaque revision suitable for a later conditional mutation. Backends must not expose a revision for a partial page. An offset equal to the current file size returns an empty page; an offset beyond it raises WorkspaceReadOffsetError.

bounded_read_limit (method)

RunnerBoundWorkspace.bounded_read_limit(max_bytes: int) -> int

Resolve a hard read ceiling without loosening backend defaults.

Return a positive integer no greater than max_bytes. Backends with their own finite default read limit must return the smaller value; backends without one return max_bytes. Callers use this before read_bytes when a safety bound must compose with backend policy.

write_bytes (method)

async RunnerBoundWorkspace.write_bytes(path: str, content: bytes) -> None

Write a file into the workspace.

delete (method)

async RunnerBoundWorkspace.delete(path: str) -> None

Delete a file from the workspace if it exists.

create_bytes (method)

async RunnerBoundWorkspace.create_bytes(path: str, content: bytes) -> WorkspaceMutationResult

Create a missing file without overwriting an existing path.

replace_bytes (method)

async RunnerBoundWorkspace.replace_bytes(path: str, content: bytes, *, expected_revision: str) -> WorkspaceMutationResult

Replace a file only when its current opaque revision matches.

delete_if_revision (method)

async RunnerBoundWorkspace.delete_if_revision(path: str, *, expected_revision: str) -> WorkspaceMutationResult

Delete a file only when its current opaque revision matches.

require_absent (method)

async RunnerBoundWorkspace.require_absent(path: str) -> None

Authoritatively preflight that one workspace path is absent.

This is an optional compatibility capability. Implementations used by the structured patch tool must override it; the later create or move still retains its own conditional absence precondition because this preflight is not a lock.

move_if_revision (method)

async RunnerBoundWorkspace.move_if_revision(source_path: str, destination_path: str, *, expected_source_revision: str, require_destination_absent: bool=True) -> WorkspaceMoveResult

Conditionally move one regular file within this workspace.

Implementations must never overwrite the destination and must report their actual move fidelity. Unsupported adapters fail explicitly; they never fall back to shell execution or an unlabelled copy/delete pair.

list (method)

async RunnerBoundWorkspace.list(pattern: str='**/*', *, limit: int | None=None) -> WorkspaceListResult

List files in the workspace.

Every backend must match pattern against workspace-relative POSIX file paths with the normative semantics of matches_list_pattern.

list_git_entries (method)

async RunnerBoundWorkspace.list_git_entries(*, limit: int) -> WorkspaceGitEntryListResult

List regular files and symlinks without following either entry type.

resource_key (property)

RunnerBoundWorkspace.resource_key() -> tuple[object, ...] | None

Stable, hashable token identifying the underlying resource (filesystem/sandbox area) this workspace reads and writes, so callers can tell whether two Workspace objects point at the SAME place.

Returns None when identity cannot be determined. SyncBinding then refuses to bind rather than risk clearing a target that is actually the source; override this in a custom Workspace to return a stable identity token and enable that safety check.

tar_copy_policy_identity (method)

RunnerBoundWorkspace.tar_copy_policy_identity() -> tuple[object, ...] | None

Return an exact, content-free identity for tar copy security policy.

SyncBinding only shares a sealed archive when both adapters opt in with stable identities. Implementations must change the identity whenever path exclusions, archive parsing, or extraction policy changes. None conservatively disables sharing for this adapter.

bounded_tar_stream_reader (method)

RunnerBoundWorkspace.bounded_tar_stream_reader() -> BoundedTarStreamReader | None

Return an explicit bounded streaming-tar reader capability, if any.

tar_stream_writer (method)

RunnerBoundWorkspace.tar_stream_writer() -> TarStreamWriter | None

Return an explicit streaming-tar writer capability, if any.

branch_capabilities (method)

RunnerBoundWorkspace.branch_capabilities() -> WorkspaceBranchCapabilities

Return explicit workspace-branch guarantees for this adapter.

Unsupported is the compatibility default. Capability must be declared by the exact adapter instance and is never inferred from adjacent filesystem, snapshot, runner, sync, reconnect, or Git behavior.

branch_lifecycle_summary (method)

RunnerBoundWorkspace.branch_lifecycle_summary() -> WorkspaceBranchLifecycleSummary

Return bounded lifecycle states for branch handles attached here.

create_branch (method)

async RunnerBoundWorkspace.create_branch(request: WorkspaceBranchRequest) -> WorkspaceBranchCreationResult

Create an isolated bounded workspace branch when supported.

Branching is an optional capability rather than an abstract workspace requirement. Backends that do not implement the complete isolation and publication contract return a typed unsupported result.

is_bound_to_runner (method)

RunnerBoundWorkspace.is_bound_to_runner(runner: Runner) -> bool

Return whether this workspace targets runner exactly.

runner_cwd (property)

RunnerBoundWorkspace.runner_cwd() -> str

Absolute runner path that this workspace exposes.

bound_runner_resource_key (property)

RunnerBoundWorkspace.bound_runner_resource_key() -> tuple[object, ...] | None

Stable runner resource identity used by workspace operations.

Release source · Runtime guarantees · Capability guides

`TarStreamReadResult`

from cayu.workspaces import TarStreamReadResult

Content-free accounting for one completed streamed tar export.

Decorators: @dataclass(frozen=True, slots=True).

Fields and attributes

Name	Type	Default
`archive_bytes`	`int`	`required`

Release source · Runtime guarantees · Capability guides

`TarStreamWriter`

from cayu.workspaces import TarStreamWriter

Nominal capability for extracting a caller-validated tar byte stream.

Bases: ABC.

write_tar_stream (method)

async TarStreamWriter.write_tar_stream(source: BinaryIO, *, archive_bytes: int) -> None

Consume exactly archive_bytes bytes without receiving a host path.

Release source · Runtime guarantees · Capability guides

`TarWriter`

from cayu.workspaces import TarWriter

Nominal capability for writing a caller-validated bulk tar archive.

Bases: ABC.

write_tar_bytes (method)

async TarWriter.write_tar_bytes(data: bytes) -> None

Extract caller-validated uncompressed tar data into this workspace.

Release source · Runtime guarantees · Capability guides

`translate_list_pattern`

from cayu.workspaces import translate_list_pattern

Translate a Workspace list() pattern into an anchored regular expression.

This defines the one normative matching semantics shared by every Workspace backend, applied to a file’s full workspace-relative POSIX path:

The pattern is anchored at both ends (*.txt does NOT match nested/a.txt).
* matches any run of characters within one path segment, ? matches one character within a segment, and [...]/[!...] character classes match one character within a segment; none of them cross /.
** as a whole segment matches zero or more directories when more pattern follows (**/*.txt matches a.txt and d/a.txt), and matches any remaining path when it is the final segment.
Empty segments and . segments in the pattern are ignored.

translate_list_pattern(pattern: str) -> str

Release source · Runtime guarantees · Capability guides

`validate_list_pattern`

from cayu.workspaces import validate_list_pattern

Validate a Workspace list() pattern shared by every backend.

validate_list_pattern(pattern: str) -> str

Release source · Runtime guarantees · Capability guides

`Workspace`

from cayu.workspaces import Workspace

Filesystem/artifact area an agent can work in.

Bases: ABC.

Fields and attributes

Name	Type	Default
`id`	`str`	`required`

reference_binding (method)

Workspace.reference_binding() -> WorkspaceReferenceBinding

Bind application data to this workspace incarnation.

read_bytes (method)

async Workspace.read_bytes(path: str, *, offset: int=0, max_bytes: int | None=None) -> WorkspaceReadResult

Read a byte page from the workspace.

bounded_read_limit (method)

Workspace.bounded_read_limit(max_bytes: int) -> int

Resolve a hard read ceiling without loosening backend defaults.

write_bytes (method)

async Workspace.write_bytes(path: str, content: bytes) -> None

Write a file into the workspace.

delete (method)

async Workspace.delete(path: str) -> None

Delete a file from the workspace if it exists.

create_bytes (method)

async Workspace.create_bytes(path: str, content: bytes) -> WorkspaceMutationResult

Create a missing file without overwriting an existing path.

replace_bytes (method)

async Workspace.replace_bytes(path: str, content: bytes, *, expected_revision: str) -> WorkspaceMutationResult

Replace a file only when its current opaque revision matches.

delete_if_revision (method)

async Workspace.delete_if_revision(path: str, *, expected_revision: str) -> WorkspaceMutationResult

Delete a file only when its current opaque revision matches.

require_absent (method)

async Workspace.require_absent(path: str) -> None

Authoritatively preflight that one workspace path is absent.

move_if_revision (method)

async Workspace.move_if_revision(source_path: str, destination_path: str, *, expected_source_revision: str, require_destination_absent: bool=True) -> WorkspaceMoveResult

Conditionally move one regular file within this workspace.

list (method)

async Workspace.list(pattern: str='**/*', *, limit: int | None=None) -> WorkspaceListResult

List files in the workspace.

Every backend must match pattern against workspace-relative POSIX file paths with the normative semantics of matches_list_pattern.

list_git_entries (method)

async Workspace.list_git_entries(*, limit: int) -> WorkspaceGitEntryListResult

List regular files and symlinks without following either entry type.

resource_key (property)

Workspace.resource_key() -> tuple[object, ...] | None

Stable, hashable token identifying the underlying resource (filesystem/sandbox area) this workspace reads and writes, so callers can tell whether two Workspace objects point at the SAME place.

tar_copy_policy_identity (method)

Workspace.tar_copy_policy_identity() -> tuple[object, ...] | None

Return an exact, content-free identity for tar copy security policy.

bounded_tar_stream_reader (method)

Workspace.bounded_tar_stream_reader() -> BoundedTarStreamReader | None

Return an explicit bounded streaming-tar reader capability, if any.

tar_stream_writer (method)

Workspace.tar_stream_writer() -> TarStreamWriter | None

Return an explicit streaming-tar writer capability, if any.

branch_capabilities (method)

Workspace.branch_capabilities() -> WorkspaceBranchCapabilities

Return explicit workspace-branch guarantees for this adapter.

Unsupported is the compatibility default. Capability must be declared by the exact adapter instance and is never inferred from adjacent filesystem, snapshot, runner, sync, reconnect, or Git behavior.

branch_lifecycle_summary (method)

Workspace.branch_lifecycle_summary() -> WorkspaceBranchLifecycleSummary

Return bounded lifecycle states for branch handles attached here.

create_branch (method)

async Workspace.create_branch(request: WorkspaceBranchRequest) -> WorkspaceBranchCreationResult

Create an isolated bounded workspace branch when supported.

Branching is an optional capability rather than an abstract workspace requirement. Backends that do not implement the complete isolation and publication contract return a typed unsupported result.

Release source · Runtime guarantees · Capability guides

`WorkspaceDirectoryPruner`

from cayu.workspaces import WorkspaceDirectoryPruner

Remove only directory trees with no files, links, or excluded paths.

Bases: ABC.

prune_empty_directories (method)

async WorkspaceDirectoryPruner.prune_empty_directories(path: str, *, max_directories: int) -> None

Remove an obstructing empty tree, bounded by max_directories.

Missing paths and regular files are unchanged. Reject links, special files, excluded descendants, and trees exceeding the bound. Callers must hold exclusive writer isolation throughout restoration.

Release source · Runtime guarantees · Capability guides

`WorkspaceGitEntry`

from cayu.workspaces import WorkspaceGitEntry

One Git-significant workspace entry observed without following symlinks.

Decorators: @dataclass(frozen=True).

Fields and attributes

Name	Type	Default
`path`	`str`	`required`
`git_mode`	`WorkspaceGitEntryMode`	`required`
`symlink_target_sha256`	`str \| None`	`None`
`symlink_target_bytes`	`int \| None`	`None`

Release source · Runtime guarantees · Capability guides

`WorkspaceGitEntryListResult`

from cayu.workspaces import WorkspaceGitEntryListResult

Bounded complete-or-truncated Git-significant entry observation.

Decorators: @dataclass(frozen=True).

Fields and attributes

Name	Type	Default
`entries`	`tuple[WorkspaceGitEntry, ...]`	`required`
`total_count`	`int`	`required`
`truncated`	`bool`	`False`

Release source · Runtime guarantees · Capability guides

`WorkspaceGitEntryObservationUnsupportedError`

from cayu.workspaces import WorkspaceGitEntryObservationUnsupportedError

A workspace cannot enumerate every Git-significant entry without following links.

Bases: RuntimeError.

This type’s implementation and inherited behavior are available in the linked release source.

Release source · Runtime guarantees · Capability guides

`WorkspaceGitMode`

from cayu.workspaces import WorkspaceGitMode

WorkspaceGitMode = Literal['100644', '100755']

Release source · Runtime guarantees · Capability guides

`WorkspaceGitModeMismatchError`

from cayu.workspaces import WorkspaceGitModeMismatchError

A conditional workspace mutation observed a different Git file mode.

Bases: RuntimeError.

init (method)

WorkspaceGitModeMismatchError.__init__(expected_git_mode: WorkspaceGitMode, actual_git_mode: WorkspaceGitMode) -> None

Release source · Runtime guarantees · Capability guides

`WorkspaceGitModeMutator`

from cayu.workspaces import WorkspaceGitModeMutator

Nominal capability for exact Git executable-mode mutations.

Bases: ABC.

write_bytes_with_git_mode (method)

async WorkspaceGitModeMutator.write_bytes_with_git_mode(path: str, content: bytes, *, git_mode: WorkspaceGitMode) -> None

Write exact bytes and normalize the file to one Git regular-file mode.

create_bytes_with_git_mode (method)

async WorkspaceGitModeMutator.create_bytes_with_git_mode(path: str, content: bytes, *, git_mode: WorkspaceGitMode) -> WorkspaceMutationResult

Create a missing file with an exact Git regular-file mode.

replace_bytes_with_git_mode (method)

async WorkspaceGitModeMutator.replace_bytes_with_git_mode(path: str, content: bytes, *, expected_revision: str, expected_git_mode: WorkspaceGitMode, git_mode: WorkspaceGitMode) -> WorkspaceMutationResult

Conditionally replace bytes and mode against both prior authorities.

Release source · Runtime guarantees · Capability guides

`WorkspaceListResult`

from cayu.workspaces import WorkspaceListResult

Decorators: @dataclass(frozen=True).

Fields and attributes

Name	Type	Default
`paths`	`tuple[str, ...]`	`required`
`total_count`	`int \| None`	`required`
`truncated`	`bool`	`False`

Release source · Runtime guarantees · Capability guides

`WorkspaceMoveAmbiguousError`

from cayu.workspaces import WorkspaceMoveAmbiguousError

A move may have created its destination before settlement became uncertain.

Bases: RuntimeError.

init (method)

WorkspaceMoveAmbiguousError.__init__(result: WorkspaceMoveResult | None=None) -> None

Release source · Runtime guarantees · Capability guides

`WorkspaceMoveFidelity`

from cayu.workspaces import WorkspaceMoveFidelity

WorkspaceMoveFidelity = Literal['atomic_rename', 'link_unlink']

Release source · Runtime guarantees · Capability guides

`WorkspaceMoveResult`

from cayu.workspaces import WorkspaceMoveResult

Evidence for one conditional, no-overwrite workspace file move.

link_unlink is intentionally distinct from atomic_rename: the destination link is created conditionally before the source name is removed, so a backend using that portable implementation must not claim a single atomic rename.

Decorators: @dataclass(frozen=True).

Fields and attributes

Name	Type	Default
`source_before_revision`	`str`	`required`
`source_after_revision`	`None`	`required`
`destination_before_revision`	`None`	`required`
`destination_after_revision`	`str`	`required`
`source_before_sha256`	`str`	`required`
`source_after_sha256`	`None`	`required`
`destination_before_sha256`	`None`	`required`
`destination_after_sha256`	`str`	`required`
`source_before_bytes`	`int`	`required`
`source_after_bytes`	`None`	`required`
`destination_before_bytes`	`None`	`required`
`destination_after_bytes`	`int`	`required`
`fidelity`	`WorkspaceMoveFidelity`	`required`

Release source · Runtime guarantees · Capability guides

`WorkspaceMoveUnsupportedError`

from cayu.workspaces import WorkspaceMoveUnsupportedError

A workspace cannot provide a conditional same-workspace regular-file move.

Bases: RuntimeError.

This type’s implementation and inherited behavior are available in the linked release source.

Release source · Runtime guarantees · Capability guides

`WorkspaceMutationResult`

from cayu.workspaces import WorkspaceMutationResult

Decorators: @dataclass(frozen=True).

Fields and attributes

Name	Type	Default
`operation`	`WorkspaceMutationOperation`	`required`
`before_revision`	`str \| None`	`required`
`after_revision`	`str \| None`	`required`
`before_sha256`	`str \| None`	`required`
`after_sha256`	`str \| None`	`required`
`before_bytes`	`int \| None`	`required`
`after_bytes`	`int \| None`	`required`

Release source · Runtime guarantees · Capability guides

`WorkspacePreconditionUnsupportedError`

from cayu.workspaces import WorkspacePreconditionUnsupportedError

A workspace cannot authoritatively evaluate a required precondition.

Bases: RuntimeError.

This type’s implementation and inherited behavior are available in the linked release source.

Release source · Runtime guarantees · Capability guides

`WorkspaceReadOffsetError`

from cayu.workspaces import WorkspaceReadOffsetError

A workspace read requested a byte offset beyond the current file size.

Bases: ValueError.

init (method)

WorkspaceReadOffsetError.__init__(offset: int, total_bytes: int) -> None

Release source · Runtime guarantees · Capability guides

`WorkspaceReadResult`

from cayu.workspaces import WorkspaceReadResult

Decorators: @dataclass(frozen=True).

Fields and attributes

Name	Type	Default
`content`	`bytes`	`required`
`total_bytes`	`int`	`required`
`truncated`	`bool`	`False`
`offset`	`int`	`0`
`revision`	`str \| None`	`None`
`sha256`	`str \| None`	`None`
`git_mode`	`WorkspaceGitMode \| None`	`None`
`source_bytes_read`	`int \| None`	`None`
`redaction_truncated`	`bool`	`False`

next_offset (property)

WorkspaceReadResult.next_offset() -> int | None

Release source · Runtime guarantees · Capability guides

`WorkspaceRevisionMismatchError`

from cayu.workspaces import WorkspaceRevisionMismatchError

A conditional workspace mutation observed a different current revision.

Bases: RuntimeError.

init (method)

WorkspaceRevisionMismatchError.__init__(expected_revision: str, actual_revision: str) -> None

Release source · Runtime guarantees · Capability guides