Skip to content

archive

archive

Archived-task queries — reading the immutable snapshots _archive_task leaves behind on deletion. Writing the archive is part of the deletion choreography and lives in terok.lib.orchestration.tasks.lifecycle.

__all__ = ['ArchivedTask', 'list_archived_tasks', 'task_archive_list', 'task_archive_logs'] module-attribute

ArchivedTask(archive_dir, archived_at, task_id, name, mode, exit_code) dataclass

Metadata snapshot of an archived (deleted) task.

archive_dir instance-attribute

archived_at instance-attribute

task_id instance-attribute

name instance-attribute

mode instance-attribute

exit_code instance-attribute

list_archived_tasks(project_name)

Return archived tasks for project_name, sorted newest-first.

Source code in src/terok/lib/orchestration/tasks/archive.py 
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
def list_archived_tasks(project_name: str) -> list[ArchivedTask]:
    """Return archived tasks for *project_name*, sorted newest-first."""
    archive_root = tasks_archive_dir(project_name)
    if not archive_root.is_dir():
        return []
    results: list[ArchivedTask] = []
    for entry in sorted(archive_root.iterdir(), reverse=True):
        if not entry.is_dir():
            continue
        meta = _load_archived_task_meta(entry)
        if meta is None:
            continue
        # Parse archive timestamp from directory name: <timestamp>_<task_id>[_<name>]
        parts = entry.name.split("_", 2)
        archived_at = parts[0] if parts else entry.name
        results.append(
            ArchivedTask(
                archive_dir=entry,
                archived_at=archived_at,
                task_id=str(meta.get("task_id", "")),
                name=meta.get("name", ""),
                mode=meta.get("mode"),
                exit_code=meta.get("exit_code"),
            )
        )
    return results

task_archive_list(project_name)

Print archived tasks for project_name.

Source code in src/terok/lib/orchestration/tasks/archive.py 
68
69
70
71
72
73
74
75
76
77
78
79
80
81
def task_archive_list(project_name: str) -> None:
    """Print archived tasks for *project_name*."""
    archived = list_archived_tasks(project_name)
    if not archived:
        print("No archived tasks found")
        return
    for a in archived:
        extra = []
        if a.mode:
            extra.append(f"mode={a.mode}")
        if a.exit_code is not None:
            extra.append(f"exit={a.exit_code}")
        extra_s = f" [{'; '.join(extra)}]" if extra else ""
        print(f"- {a.archived_at} #{a.task_id}: {a.name}{extra_s}")

task_archive_logs(project_name, archive_id)

Return the log file path for an archived task identified by archive_id.

archive_id is matched against archive directory names (prefix match). Returns the log file path if found, or None.

Source code in src/terok/lib/orchestration/tasks/archive.py 
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
def task_archive_logs(project_name: str, archive_id: str) -> Path | None:
    """Return the log file path for an archived task identified by *archive_id*.

    *archive_id* is matched against archive directory names (prefix match).
    Returns the log file path if found, or ``None``.
    """
    archive_root = tasks_archive_dir(project_name)
    if not archive_root.is_dir():
        return None
    for entry in sorted(archive_root.iterdir(), reverse=True):
        if entry.is_dir() and entry.name.startswith(archive_id):
            log_file = entry / "logs" / "container.log"
            if log_file.is_file():
                return log_file
    return None