Skip to content

Storage

storage

Filesystem storage queries for agent-owned directories.

Every task leaves a footprint on the host: a workspace, agent config files, and shared config mounts that survive across containers. This module measures those footprints so the orchestrator can report them.

TaskStorageInfo(task_id, workspace_bytes, agent_config_bytes) dataclass

Disk usage snapshot for a single task's host directories.

task_id instance-attribute

workspace_bytes instance-attribute

agent_config_bytes instance-attribute

total_bytes property

Combined footprint of workspace and agent config.

measure(task_dir) classmethod

Measure a single task's disk footprint.

Expects the standard layout: <task_dir>/workspace-dangerous/ for the agent-writable code, <task_dir>/agent-config/ for per-task configuration.

Source code in src/terok_executor/storage.py 
33
34
35
36
37
38
39
40
41
42
43
44
45
@classmethod
def measure(cls, task_dir: Path) -> TaskStorageInfo:
    """Measure a single task's disk footprint.

    Expects the standard layout: ``<task_dir>/workspace-dangerous/``
    for the agent-writable code, ``<task_dir>/agent-config/`` for
    per-task configuration.
    """
    return cls(
        task_id=task_dir.name,
        workspace_bytes=_dir_bytes(task_dir / "workspace-dangerous"),
        agent_config_bytes=_dir_bytes(task_dir / "agent-config"),
    )

measure_all(tasks_root) classmethod

Measure every task under tasks_root, sorted by task ID.

Source code in src/terok_executor/storage.py 
47
48
49
50
51
52
53
54
55
@classmethod
def measure_all(cls, tasks_root: Path) -> list[TaskStorageInfo]:
    """Measure every task under *tasks_root*, sorted by task ID."""
    if not tasks_root.is_dir():
        return []
    return sorted(
        (cls.measure(d) for d in tasks_root.iterdir() if d.is_dir()),
        key=lambda t: t.task_id,
    )

SharedMountStorageInfo(name, label, bytes) dataclass

Disk usage for one shared config mount directory.

name instance-attribute

label instance-attribute

bytes instance-attribute

measure_all(mounts_base=None) classmethod

Measure each shared config mount directory.

Labels come from the agent roster when available, falling back to a title-cased version of the directory name.

Source code in src/terok_executor/storage.py 
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
@classmethod
def measure_all(cls, mounts_base: Path | None = None) -> list[SharedMountStorageInfo]:
    """Measure each shared config mount directory.

    Labels come from the agent roster when available, falling back to
    a title-cased version of the directory name.
    """
    base = mounts_base or mounts_dir()
    if not base.is_dir():
        return []

    roster_mounts = AgentRoster.shared().mounts
    return sorted(
        (
            cls(
                name=d.name,
                label=_mount_label(d.name, roster_mounts),
                bytes=_dir_bytes(d),
            )
            for d in base.iterdir()
            if d.is_dir()
        ),
        key=lambda m: m.name,
    )