project

project

Project entities, lifecycle, panic, SSH provisioning — public API surface.

Re-export catalog for everything project-shaped. Sources: terok.lib.core.projects for the pure ProjectConfig value type and discovery helpers; terok.lib.domain.project for the rich Project aggregate and lifecycle; terok.lib.domain.project_state for infrastructure-state queries; terok.lib.domain.image_cleanup for image listing & cleanup; terok.lib.domain.panic for cross-project lockdown; terok.lib.domain.ssh for the SSH provisioning workflow; and terok.lib.domain.wizards.new_project for the new-project wizard primitives that CLI prompts and TUI screens share.

AGENTS_QUESTION = Question(key='agents', kind='multichoice', prompt='Select agents to install', help="Which AI coding agents to bake into this project's image, overriding the global default. Pick 'All agents' to inherit future additions, or enumerate specific agents to freeze the set.", required=True, choices_loader=_load_agent_choices, validate=_validate_agents) module-attribute

QUESTIONS = (Question(key='security_class', kind='choice', prompt='Select security mode', choices=(tuple(SECURITY_CLASSES)), required=True), Question(key='base', kind='choice', prompt='Select base image', choices=(tuple(BASES)), required=True), Question(key='project_id', kind='text', prompt='Project ID', required=True, transform=_slugify_project_id, validate=_validate_project_id, placeholder='lowercase; letters, digits, hyphens, underscores'), Question(key='upstream_url', kind='text', prompt='Upstream git URL', help='Leave empty for a local-only project (no remote).', placeholder='git@github.com:org/repo.git or https://…', default_visible=True), Question(key='default_branch', kind='text', prompt='Default branch', help="Leave empty to use the remote's default (or ``main`` when no remote).", placeholder='main', default_visible=True), Question(key='user_snippet', kind='editor', prompt='Custom image snippet', help='Optional Dockerfile fragment appended to the project image. Use for extra packages, env vars, or setup commands.', default_visible=True)) module-attribute

__all__ = ['AGENTS_QUESTION', 'BrokenProject', 'DeleteProjectResult', 'Project', 'ProjectConfig', 'QUESTIONS', 'Question', 'cleanup_images', 'delete_project', 'derive_project', 'discover_projects', 'execute_panic', 'find_orphaned_images', 'find_projects_sharing_gate', 'format_panic_report', 'get_project', 'list_images', 'list_projects', 'load_project', 'panic_stop_containers', 'project_image_exists', 'remove_images', 'render_project_yaml', 'require_project_exists', 'set_project_image_agents', 'summarize_ssh_init', 'validate_answer', 'write_project_yaml'] module-attribute

BrokenProject(id, config_path, error) dataclass

A project directory whose project.yml failed to load.

Carries just enough context for the TUI to render a row and show the validation error in the details pane, without forcing callers to re-run the failing load_project to rediscover the message.

id instance-attribute

config_path instance-attribute

error instance-attribute

ProjectConfig

Bases: BaseModel

Resolved project configuration loaded from project.yml.

Pure value object — holds configuration fields with no behavior beyond computed paths. The rich domain object Project wraps this and provides behavior.

model_config = ConfigDict(frozen=True) class-attribute instance-attribute

id instance-attribute

security_class instance-attribute

isolation = 'shared' class-attribute instance-attribute

upstream_url instance-attribute

default_branch instance-attribute

root instance-attribute

tasks_root instance-attribute

gate_path instance-attribute

gate_enabled = True class-attribute instance-attribute

staging_root instance-attribute

ssh_use_personal = False class-attribute instance-attribute

Opt in to the user's ~/.ssh keys for host-side gate-sync (default off).

expose_external_remote = False class-attribute instance-attribute

human_name = None class-attribute instance-attribute

human_email = None class-attribute instance-attribute

git_authorship = 'agent-human' class-attribute instance-attribute

upstream_polling_enabled = True class-attribute instance-attribute

upstream_polling_interval_minutes = 5 class-attribute instance-attribute

auto_sync_enabled = False class-attribute instance-attribute

auto_sync_branches = Field(default_factory=list) class-attribute instance-attribute

default_agent = None class-attribute instance-attribute

default_login = None class-attribute instance-attribute

agent_config = Field(default_factory=dict) class-attribute instance-attribute

shutdown_timeout = 10 class-attribute instance-attribute

memory = None class-attribute instance-attribute

Podman --memory value from run.memory in project.yml.

cpus = None class-attribute instance-attribute

Podman --cpus value from run.cpus in project.yml.

nested_containers = False class-attribute instance-attribute

Project runs podman/docker inside its container (see run.nested_containers).

runtime = None class-attribute instance-attribute

OCI runtime selector from run.runtime.

None (default) means "use the global default", which itself falls through to "crun" — the OCI runtime podman drives by default on every supported distro. "krun" selects KVM-microVM isolation; gated on the global experimental: true flag at runtime selection time so a typo never silently boots the experimental backend.

Sizing reuses the standard memory / cpus knobs — podman writes them into the OCI spec and the runtime reads them there; no krun-specific knob.

timezone = None class-attribute instance-attribute

IANA timezone for task containers (from run.timezone).

None lets terok-executor fall back to the host's timezone; pass an explicit string ("UTC", "Europe/Prague") to override — including to pin containers to UTC for reproducible runs.

task_name_categories = None class-attribute instance-attribute

shield_drop_on_task_run = True class-attribute instance-attribute

shield_on_task_restart = 'retain' class-attribute instance-attribute

hook_pre_start = None class-attribute instance-attribute

hook_post_start = None class-attribute instance-attribute

hook_post_ready = None class-attribute instance-attribute

hook_post_stop = None class-attribute instance-attribute

base_image = 'ubuntu:24.04' class-attribute instance-attribute

family = None class-attribute instance-attribute

Package family override for L0/L1 builds.

None lets terok-executor auto-detect from base_image; set explicitly when the auto-detect allowlist doesn't recognise the image (rocky, alma, suse, …).

agents = 'all' class-attribute instance-attribute

Comma-separated roster entries to install in L1 (or "all").

snippet_inline = None class-attribute instance-attribute

snippet_file = None class-attribute instance-attribute

shared_dir = None class-attribute instance-attribute

is_sealed property

Whether this project uses sealed isolation (zero bind mounts).

presets_dir property

Directory for preset config files for this project.

DeleteProjectResult

Bases: TypedDict

Result of a project deletion.

deleted instance-attribute

skipped instance-attribute

archive instance-attribute

Project(config)

Rich project object — DDD Aggregate Root.

The primary domain object that callers interact with. Wraps a ProjectConfig value object and exposes all project-scoped operations through a natural OOP interface::

project = get_project("myproj")
task = project.create_task(name="fix-bug")
task.run_cli()
task.stop()
project.gate.sync()

Identity is based on project.id — two Project instances with the same ID compare equal and hash identically, so they work correctly in sets and dicts.

Subsystem access (gate, ssh, agents) uses lazy initialization: the service objects are created on first property access rather than at construction time. This avoids unnecessary I/O when only a subset of functionality is needed. Uses __slots__ for memory efficiency; cached_property is not available because it requires __dict__.

Obtain via get_project or list_projects.

Initialize with a resolved project configuration.

Source code in src/terok/lib/domain/project.py

def __init__(self, config: ProjectConfig) -> None:
    """Initialize with a resolved project configuration."""
    self._config = config
    self._gate: GitGate | None = None
    self._ssh: SSHManager | None = None
    self._agents: AgentManager | None = None

__slots__ = ('_config', '_gate', '_ssh', '_agents') class-attribute instance-attribute

id property

Return the project ID.

config property

Return the underlying configuration value object.

security_class property

Return the project's security class ('online' or 'gatekeeping').

tasks property

All tasks in this project — convenience for unfiltered iteration.

gate property

Return the project-scoped git gate manager (lazy-initialized).

ssh property

Return the project-scoped SSH manager (lazy-initialized).

needs_ssh_key_registration property

Return True when the upstream is SSH-scheme so a deploy key must be added.

Shared predicate used by the CLI pause helper and the TUI wizard's mid-flow "continue" gate — keeps the rule (SSH URLs need registration, HTTPS and no-upstream projects don't) in one place.

agents property

Return the project-scoped agent configuration manager (lazy-initialized).

__eq__(other)

Two projects are equal iff they share the same ID.

Source code in src/terok/lib/domain/project.py

def __eq__(self, other: object) -> bool:
    """Two projects are equal iff they share the same ID."""
    return isinstance(other, Project) and self.id == other.id

__hash__()

Hash by project ID for use in sets and dicts.

Source code in src/terok/lib/domain/project.py

def __hash__(self) -> int:
    """Hash by project ID for use in sets and dicts."""
    return hash(self.id)

create_task(*, name=None)

Create a new task and return a rich Task entity.

Source code in src/terok/lib/domain/project.py

def create_task(self, *, name: str | None = None) -> Task:
    """Create a new task and return a rich Task entity."""
    task_id = task_new(self._config.id, name=name)
    meta = get_task_meta(self._config.id, task_id)
    return Task(self._config, meta)

get_task(task_id)

Return a rich Task entity for an existing task.

Source code in src/terok/lib/domain/project.py

def get_task(self, task_id: str) -> Task:
    """Return a rich Task entity for an existing task."""
    meta = get_task_meta(self._config.id, task_id)
    return Task(self._config, meta)

list_tasks(*, status=None, mode=None)

Return all tasks, optionally filtered by status or mode.

Source code in src/terok/lib/domain/project.py

def list_tasks(self, *, status: str | None = None, mode: str | None = None) -> list[Task]:
    """Return all tasks, optionally filtered by status or mode."""
    metas = get_tasks(self._config.id)
    if mode:
        metas = [m for m in metas if m.mode == mode]
    # Hydrate live container state so status filtering is accurate
    live_states = get_all_task_states(self._config.id, metas)
    for m in metas:
        m.container_state = live_states.get(m.task_id)
    if status:
        metas = [m for m in metas if m.status == status]
    return [Task(self._config, m) for m in metas]

acp_endpoints()

Return one ACPEndpoint per running task.

Cheap discovery surface — walks running tasks, classifies each endpoint as active (daemon up, socket bound), ready (task running with at least one authed agent, daemon would spawn on first connect), or unsupported (no agents authed for this task's image; connect would fail).

No probing, no socket traffic — one credential-DB read for the whole listing, one Sandbox instance shared across tasks, and image-label lookups memoised by image-id (most tasks share an image). terok acp list and the TUI panel share this entry point.

Source code in src/terok/lib/domain/project.py

def acp_endpoints(self) -> list[ACPEndpoint]:
    """Return one [`ACPEndpoint`][terok.lib.domain.project.ACPEndpoint] per running task.

    Cheap discovery surface — walks running tasks, classifies each
    endpoint as ``active`` (daemon up, socket bound), ``ready``
    (task running with at least one authed agent, daemon would
    spawn on first connect), or ``unsupported`` (no agents authed
    for this task's image; connect would fail).

    No probing, no socket traffic — one credential-DB read for
    the whole listing, one ``Sandbox`` instance shared across
    tasks, and image-label lookups memoised by image-id (most
    tasks share an image).  ``terok acp list`` and the TUI panel
    share this entry point.
    """
    running = self.list_tasks(status="running")
    if not running:
        return []
    from terok.lib.integrations.sandbox import Sandbox

    # One DB read + one Sandbox + per-image label cache for the
    # whole listing.  Auth is global today — same set for every task.
    authed = set(list_authenticated_agents())
    sandbox = Sandbox(config=make_sandbox_config())
    label_cache: dict[str, set[str]] = {}
    out: list[ACPEndpoint] = []
    for task in running:
        sock = acp_socket_path(self._config.id, task.id)
        sock_exists = sock.exists()
        bound = _read_bound_agent(self._config.id, task.id) if sock_exists else None
        if sock_exists:
            status = ACPEndpointStatus.ACTIVE
        elif _task_has_any_authed_agent(
            self._config.id, task, authed, sandbox=sandbox, label_cache=label_cache
        ):
            status = ACPEndpointStatus.READY
        else:
            status = ACPEndpointStatus.UNSUPPORTED
        out.append(
            ACPEndpoint(
                project_id=self._config.id,
                task_id=task.id,
                socket_path=sock,
                status=status,
                bound_agent=bound,
            )
        )
    return out

run_headless(request)

Create and run a headless task atomically. Returns the Task.

Source code in src/terok/lib/domain/project.py

def run_headless(self, request: HeadlessRunRequest) -> Task:
    """Create and run a headless task atomically.  Returns the Task."""
    task_id = task_run_headless(request)
    meta = get_task_meta(self._config.id, task_id)
    return Task(self._config, meta)

followup_headless(task_id, prompt, follow=True)

Send a follow-up prompt to a completed headless task.

Source code in src/terok/lib/domain/project.py

def followup_headless(self, task_id: str, prompt: str, follow: bool = True) -> None:
    """Send a follow-up prompt to a completed headless task."""
    from ..orchestration.task_runners import task_followup_headless

    task_followup_headless(self._config.id, task_id, prompt, follow=follow)

delete()

Delete the project and all associated data.

Source code in src/terok/lib/domain/project.py

def delete(self) -> DeleteProjectResult:
    """Delete the project and all associated data."""
    return delete_project(self._config.id)

generate_dockerfiles()

Render and write Dockerfiles for this project.

Source code in src/terok/lib/domain/project.py

def generate_dockerfiles(self) -> None:
    """Render and write Dockerfiles for this project."""
    generate_dockerfiles(self._config.id)

build_images(*, include_dev=False, refresh_agents=False, full=False)

Build container images for this project.

Source code in src/terok/lib/domain/project.py

def build_images(
    self, *, include_dev: bool = False, refresh_agents: bool = False, full: bool = False
) -> None:
    """Build container images for this project."""
    build_images(
        self._config.id,
        include_dev=include_dev,
        refresh_agents=refresh_agents,
        full_rebuild=full,
    )

state(*, gate_commit_provider=None)

Return the project's infrastructure state snapshot.

gate_commit_provider is an optional callable that, given a project id, returns the last gate commit dict (or None). Used by the TUI to inject the live gate manager's last_commit lookup without reaching for it from inside the helper.

Source code in src/terok/lib/domain/project.py

def state(self, *, gate_commit_provider: Callable[[str], dict | None] | None = None) -> dict:
    """Return the project's infrastructure state snapshot.

    *gate_commit_provider* is an optional callable that, given a
    project id, returns the last gate commit dict (or ``None``).
    Used by the TUI to inject the live gate manager's ``last_commit``
    lookup without reaching for it from inside the helper.
    """
    from .project_state import get_project_state

    return get_project_state(
        self._config.id, gate_commit_provider=gate_commit_provider, project=self._config
    )

storage_detail()

Return a detailed view of this project's on-disk footprint.

Source code in src/terok/lib/domain/project.py

def storage_detail(self) -> ProjectDetail:
    """Return a detailed view of this project's on-disk footprint."""
    from .storage import get_project_storage_detail

    return get_project_storage_detail(self._config.id)

provision_ssh_key(*, key_type='ed25519', comment=None, force=False)

Mint a vault-backed keypair and bind it to this project's scope.

Opens a fresh SSHManager via the context-manager form so the credential DB closes after init, then assigns the new key_id to the project scope. Rendering the result is the caller's job — see summarize_ssh_init.

Source code in src/terok/lib/domain/project.py

def provision_ssh_key(
    self,
    *,
    key_type: str = "ed25519",
    comment: str | None = None,
    force: bool = False,
) -> SSHInitResult:
    """Mint a vault-backed keypair and bind it to this project's scope.

    Opens a fresh
    [`SSHManager`][terok_sandbox.SSHManager] via the context-manager
    form so the credential DB closes after init, then assigns the
    new ``key_id`` to the project scope.  Rendering the result is
    the caller's job — see
    [`summarize_ssh_init`][terok.lib.domain.ssh.summarize_ssh_init].
    """
    with make_ssh_manager(self._config) as ssh:
        result = ssh.init(key_type=key_type, comment=comment, force=force)
    self.register_ssh_key(result["key_id"])
    return result

register_ssh_key(key_id)

Bind an already-minted key_id to this project (idempotent).

Source code in src/terok/lib/domain/project.py

def register_ssh_key(self, key_id: int) -> None:
    """Bind an already-minted *key_id* to this project (idempotent)."""
    with vault_db() as db:
        db.assign_ssh_key(self._config.id, key_id)

pause_for_ssh_key_registration_if_needed()

Pause so the user can register the deploy key — only for SSH upstreams.

Source code in src/terok/lib/domain/project.py

def pause_for_ssh_key_registration_if_needed(self) -> None:
    """Pause so the user can register the deploy key — only for SSH upstreams."""
    if self.needs_ssh_key_registration:
        print("\n" + "=" * 60)
        print("ACTION REQUIRED: Add the public key shown above as a")
        print("deploy key (or to your SSH keys) on the git remote.")
        print("=" * 60)
        input("Press Enter once the key is registered... ")

list_presets()

Return available presets for this project.

Source code in src/terok/lib/domain/project.py

def list_presets(self) -> list[PresetInfo]:
    """Return available presets for this project."""
    return list_presets(self._config.id)

__repr__()

Return a developer-friendly string representation.

Source code in src/terok/lib/domain/project.py

def __repr__(self) -> str:
    """Return a developer-friendly string representation."""
    return f"Project(id={self.id!r}, security={self.security_class!r})"

Question(key, kind, prompt, help='', choices=(), choices_loader=None, required=False, transform=None, validate=None, placeholder='', default_visible=False) dataclass

One wizard prompt — what to ask, how to validate, what shape the answer takes.

The presenter decides the visual treatment (numbered menu vs radio buttons, input() vs Textual Input, $EDITOR vs TextArea); the declaration here drives everything else.

key instance-attribute

Name of this field in the collected-values dict.

kind instance-attribute

Shape of the input — drives which widget / prompt style a presenter uses.

prompt instance-attribute

Short one-line question, used as both CLI prompt and TUI label.

help = '' class-attribute instance-attribute

Longer explanation, rendered next to the input in the TUI; unused in CLI.

choices = () class-attribute instance-attribute

Static (value, label) pairs for kind in {"choice", "multichoice"}.

choices_loader = None class-attribute instance-attribute

Runtime resolver for choices that aren't known at import time.

Set this when the option set lives in a sibling wheel (e.g. the agent roster) and can drift between releases. When set, takes precedence over choices.

required = False class-attribute instance-attribute

Reject empty answers with "<prompt> is required."

transform = None class-attribute instance-attribute

Optional normalisation applied before validation (e.g. str.lower).

validate = None class-attribute instance-attribute

Optional validator returning an error string or None when accepted.

placeholder = '' class-attribute instance-attribute

Hint string, rendered inside the Textual Input; unused in CLI.

default_visible = False class-attribute instance-attribute

When True, CLI prompt shows "(optional)" to telegraph "Enter is fine".

resolve_choices()

Return the effective option list — runtime loader wins over static.

Called by both presenters whenever they need to render or validate a choice / multichoice question. The loader is expected to be cheap; the executor's roster lookup is itself lru_cache'd so repeated calls are free.

Source code in src/terok/lib/domain/wizards/new_project.py

def resolve_choices(self) -> tuple[tuple[str, str], ...]:
    """Return the effective option list — runtime loader wins over static.

    Called by both presenters whenever they need to render or validate
    a ``choice`` / ``multichoice`` question.  The loader is expected
    to be cheap; the executor's roster lookup is itself ``lru_cache``'d
    so repeated calls are free.
    """
    return self.choices_loader() if self.choices_loader is not None else self.choices

discover_projects()

Load every project on disk, splitting successes from config-level failures.

The broken list lets the TUI render damaged projects alongside healthy ones (issue #565) — silently hiding them turns "project vanished" into a mystery. _parse_project_yaml wraps every config error (bad YAML, schema drift, filesystem issues) in SystemExit with a human-readable message; anything else propagates as a genuine bug.

Source code in src/terok/lib/core/projects.py

def discover_projects() -> tuple[list[ProjectConfig], list[BrokenProject]]:
    """Load every project on disk, splitting successes from config-level failures.

    The broken list lets the TUI render damaged projects alongside healthy
    ones (issue #565) — silently hiding them turns "project vanished" into
    a mystery.  ``_parse_project_yaml`` wraps every config error (bad YAML,
    schema drift, filesystem issues) in ``SystemExit`` with a human-readable
    message; anything else propagates as a genuine bug.
    """
    paths_by_id = _discover_project_paths()
    valid: list[ProjectConfig] = []
    broken: list[BrokenProject] = []
    for pid in sorted(paths_by_id):
        try:
            valid.append(load_project(pid))
        except SystemExit as exc:
            msg = _sanitize_for_tty(str(exc))
            broken.append(BrokenProject(id=pid, config_path=paths_by_id[pid], error=msg))
    return valid, broken

load_project(project_id)

Load and return a fully resolved ProjectConfig from project_id.

Source code in src/terok/lib/core/projects.py

def load_project(project_id: str) -> ProjectConfig:
    """Load and return a fully resolved [`ProjectConfig`][terok.cli.commands.sickbay.ProjectConfig] from *project_id*."""
    root = _find_project_root(project_id)
    cfg_path = root / _PROJECT_YML
    if not cfg_path.is_file():
        raise SystemExit(f"Missing {_PROJECT_YML} in {root}")

    raw = _parse_project_yaml(cfg_path)

    # Git identity resolved via ConfigStack: git-global → terok-global → project.yml
    git_dict = raw.git.model_dump(exclude_none=True)
    identity_stack = ConfigStack()
    identity_stack.push(ConfigScope("git-global", None, _git_global_identity()))
    identity_stack.push(ConfigScope("terok-global", None, _validated_global_git_section()))
    identity_stack.push(ConfigScope("project", cfg_path, git_dict))
    identity = identity_stack.resolve()

    try:
        return _build_project_config(raw, identity, root, project_id)
    except ValidationError as exc:
        # Identity values come from merged sources (git config, global config,
        # project.yml).  Include provenance in the error so the user knows
        # where to look.
        sources = ", ".join(s.level for s in identity_stack.scopes if s.data)
        raise SystemExit(
            _format_validation_error(exc, cfg_path) + f"\n  (git identity merged from: {sources})"
        )

require_project_exists(project_id)

Raise SystemExit unless project_id names a known project.

Cheap stat-based check — no YAML parse, no pydantic validation. Use this in CLI entry points that want to fail before any user-visible side effect (interactive prompt, status print, image build offer). The downstream load_project call still catches malformed YAML.

Source code in src/terok/lib/core/projects.py

def require_project_exists(project_id: str) -> None:
    """Raise [`SystemExit`][SystemExit] unless *project_id* names a known project.

    Cheap stat-based check — no YAML parse, no pydantic validation.  Use
    this in CLI entry points that want to fail before any user-visible
    side effect (interactive prompt, status print, image build offer).
    The downstream [`load_project`][terok.lib.core.projects.load_project]
    call still catches malformed YAML.
    """
    _find_project_root(project_id)

set_project_image_agents(project_id, selection)

Write selection into the project's project.yml under image.agents.

Caller validates selection up-front; on success returns the project.yml path written.

Source code in src/terok/lib/core/projects.py

def set_project_image_agents(project_id: str, selection: str) -> Path:
    """Write *selection* into the project's ``project.yml`` under ``image.agents``.

    Caller validates *selection* up-front; on success returns the
    project.yml path written.
    """
    from terok.lib.integrations.sandbox import yaml_update_section

    cfg_path = _find_project_root(project_id) / _PROJECT_YML
    yaml_update_section(cfg_path, "image", {"agents": selection})
    return cfg_path

cleanup_images(dry_run=False)

Find and remove orphaned terok images in one shot.

Thin convenience over find_orphaned_images + remove_images for callers that don't need to inspect the list first.

Parameters:

Name	Type	Description	Default
dry_run	bool	If True, only report what would be removed without removing.	False

Returns:

Type	Description
CleanupResult	CleanupResult with lists of removed and failed image display names.

Source code in src/terok/lib/domain/image_cleanup.py

def cleanup_images(dry_run: bool = False) -> CleanupResult:
    """Find and remove orphaned terok images in one shot.

    Thin convenience over [`find_orphaned_images`][terok.lib.domain.image_cleanup.find_orphaned_images]
    + [`remove_images`][terok.lib.domain.image_cleanup.remove_images] for callers that
    don't need to inspect the list first.

    Args:
        dry_run: If True, only report what would be removed without removing.

    Returns:
        CleanupResult with lists of removed and failed image display names.
    """
    return remove_images(find_orphaned_images(), dry_run=dry_run)

find_orphaned_images()

Find terok images that are orphaned and safe to remove.

Orphaned images include: - Dangling images (<none>:<none>) from terok layer rebuilds - L2 project images whose project no longer exists in the config

Source code in src/terok/lib/domain/image_cleanup.py

def find_orphaned_images() -> list[ImageInfo]:
    """Find terok images that are orphaned and safe to remove.

    Orphaned images include:
    - Dangling images (``<none>:<none>``) from terok layer rebuilds
    - L2 project images whose project no longer exists in the config
    """
    known_ids = _known_project_ids()

    # Dangling images that descended from terok base layers
    dangling = _find_dangling_terok_images()

    # L2 images for projects that no longer exist (skip if discovery failed)
    orphaned_l2: list[ImageInfo] = []
    if known_ids is not None:
        all_images = list_images()
        orphaned_l2 = [
            img
            for img in all_images
            if _is_terok_l2_image(img.repository, img.tag)
            and img.project_key not in known_ids
            and _is_terok_built_image(img.image_id)
        ]

    # Combine, dedup by image ID
    seen_ids: set[str] = set()
    result: list[ImageInfo] = []
    for img in [*dangling, *orphaned_l2]:
        if img.image_id not in seen_ids:
            seen_ids.add(img.image_id)
            result.append(img)
    return result

list_images(project_id=None)

List terok-managed images, optionally filtered by project.

Parameters:

Name	Type	Description	Default
project_id	str | None	If given, only show images for this project.	None

Returns:

Type	Description
list[ImageInfo]	List of ImageInfo objects for matching images.

Source code in src/terok/lib/domain/image_cleanup.py

def list_images(project_id: str | None = None) -> list[ImageInfo]:
    """List terok-managed images, optionally filtered by project.

    Args:
        project_id: If given, only show images for this project.

    Returns:
        List of ImageInfo objects for matching images.
    """
    images: list[ImageInfo] = []
    for image in PodmanRuntime().images():
        repo = image.repository.removeprefix("localhost/")
        if not _is_terok_image(repo, image.tag):
            continue
        if project_id is not None:
            # Filter: L2 images must match the project; L0/L1 always shown
            if _is_terok_l2_image(repo, image.tag) and repo != project_id:
                continue
        images.append(ImageInfo.from_image(image))
    return images

remove_images(images, *, dry_run=False)

Remove a pre-computed set of images (or just report under dry_run).

Split out from cleanup_images so the CLI can present the orphan list, prompt for confirmation, and only then act — without paying the discovery cost twice.

Parameters:

Name	Type	Description	Default
images	Iterable[ImageInfo]	The images to remove. Iterated once.	required
dry_run	bool	If True, only report names without invoking the runtime.	False

Returns:

Type	Description
CleanupResult	CleanupResult with lists of removed and failed image display names.

Source code in src/terok/lib/domain/image_cleanup.py

def remove_images(images: Iterable[ImageInfo], *, dry_run: bool = False) -> CleanupResult:
    """Remove a pre-computed set of images (or just report under *dry_run*).

    Split out from [`cleanup_images`][terok.lib.domain.image_cleanup.cleanup_images]
    so the CLI can present the orphan list, prompt for confirmation, and only
    then act — without paying the discovery cost twice.

    Args:
        images: The images to remove.  Iterated once.
        dry_run: If True, only report names without invoking the runtime.

    Returns:
        CleanupResult with lists of removed and failed image display names.
    """
    removed: list[str] = []
    failed: list[str] = []
    for img in images:
        if dry_run:
            removed.append(img.full_name)
            continue
        try:
            if PodmanRuntime().image(img.image_id).remove():
                removed.append(img.full_name)
            else:
                failed.append(img.full_name)
        except Exception as exc:  # noqa: BLE001 — one bad image shouldn't abort the sweep
            from ..util.logging_utils import log_warning

            log_warning(f"Image cleanup failed for {img.full_name}: {exc}")
            failed.append(img.full_name)
    return CleanupResult(removed=removed, failed=failed, dry_run=dry_run)

execute_panic(*, stop_containers=False)

Execute the full panic sequence.

Discovers every running container, then raises shields, stops vault and gate — all in parallel. If stop_containers, also kills the containers afterwards (SIGKILL; they are not removed).

Source code in src/terok/lib/domain/panic.py

def execute_panic(
    *,
    stop_containers: bool = False,
) -> PanicResult:
    """Execute the full panic sequence.

    Discovers every running container, then raises shields, stops vault
    and gate — all in parallel.  If *stop_containers*, also kills the
    containers afterwards (SIGKILL; they are not removed).
    """
    result = PanicResult()
    targets = _discover_targets()
    result.total_running = len(targets)
    result.shield_bypassed = get_shield_bypass_firewall_no_protection()

    _phase1_lockdown(result, targets)
    _write_panic_lock()

    # Phase 2: optional container kill
    if stop_containers and targets:
        result.containers_stopped, result.container_stop_errors = _stop_containers(targets)

    return result

format_panic_report(result)

Format a human-readable summary of the panic result.

Source code in src/terok/lib/domain/panic.py

def format_panic_report(result: PanicResult) -> str:
    """Format a human-readable summary of the panic result."""
    lines = [
        f"Containers found: {result.total_running}",
        _format_shield_status(result),
        f"Vault: {'locked + stopped' if result.vault_stopped else 'FAILED'}",
        f"Gate:  {'stopped' if result.gate_stopped else 'FAILED'}",
    ]

    if result.containers_stopped:
        lines.append(f"Containers killed: {len(result.containers_stopped)}")

    if result.has_errors:
        lines += ["", "Errors:", *_format_errors(result)]

    return "\n".join(lines)

panic_stop_containers()

Discover and SIGKILL all running containers (Phase 2 standalone).

Source code in src/terok/lib/domain/panic.py

def panic_stop_containers() -> tuple[list[str], list[tuple[str, str]]]:
    """Discover and SIGKILL all running containers (Phase 2 standalone)."""
    return _stop_containers(_discover_targets())

delete_project(project_id)

Delete a project and all its associated data.

Removes task workspaces, task metadata, build artifacts, SSH credentials, the git gate (if not shared with other projects), and the project config directory.

Source code in src/terok/lib/domain/project.py

def delete_project(project_id: str) -> DeleteProjectResult:
    """Delete a project and all its associated data.

    Removes task workspaces, task metadata, build artifacts, SSH credentials,
    the git gate (if not shared with other projects), and the project config
    directory.
    """
    archive_path = _archive_project(project_id)
    if archive_path is None:
        raise SystemExit(
            f"Project archiving failed for '{project_id}'; aborting deletion to prevent data loss."
        )

    project = load_project(project_id)
    pid = project.id
    deleted: list[str] = []
    skipped: list[str] = []

    # 1. Stop + remove all tasks
    for task in get_tasks(pid):
        try:
            task_delete(pid, task.task_id)
        except Exception as exc:
            _logger.warning("Failed to delete task %s: %s", task.task_id, exc)

    # 2. Remove tasks root (may be user-configured path)
    _rmtree_managed(project.tasks_root, "Tasks root", deleted, skipped)

    # 3-4. Remove state dir, build artifacts, and any remaining task archives
    for d in (core_state_dir() / "projects" / pid, build_dir() / pid, archive_dir() / pid):
        if d.is_dir():
            shutil.rmtree(d)
            deleted.append(str(d))

    # 5. SSH credentials — unassign from vault; orphan keys cascade-delete.
    _unassign_vault_ssh_keys(pid, deleted, skipped)

    # 6. Git gate (skip if shared with other projects)
    sharing = find_projects_sharing_gate(project.gate_path, exclude_project=pid)
    if sharing:
        names = ", ".join(p for p, _ in sharing)
        skipped.append(f"Gate {project.gate_path} shared with: {names}")
    else:
        _rmtree_managed(project.gate_path, "Gate", deleted, skipped)

    # 7. Staging root (gatekeeping mode, may be user-configured path)
    if project.staging_root:
        _rmtree_managed(project.staging_root, "Staging root", deleted, skipped)

    # 8. Project config directory
    if project.root.is_dir():
        shutil.rmtree(project.root)
        deleted.append(str(project.root))

    return DeleteProjectResult(deleted=deleted, skipped=skipped, archive=archive_path)

derive_project(source_id, new_id)

Copy source_id's gate mirror and vault SSH assignments under new_id.

Source code in src/terok/lib/domain/project.py

def derive_project(source_id: str, new_id: str) -> Project:
    """Copy *source_id*'s gate mirror and vault SSH assignments under *new_id*."""
    _derive_project(source_id, new_id)
    _share_ssh_key_assignments(source_id, new_id)
    return Project(load_project(new_id))

find_projects_sharing_gate(gate_path, exclude_project=None)

Find all projects configured to use the same gate path.

Parameters:

Name	Type	Description	Default
gate_path	Path	The gate path to check for	required
exclude_project	str | None	Project ID to exclude from results (usually the current project)	None

Returns:

Type	Description
list[tuple[str, str | None]]	List of (project_id, upstream_url) tuples for projects sharing this gate

Source code in src/terok/lib/domain/project.py

def find_projects_sharing_gate(
    gate_path: Path, exclude_project: str | None = None
) -> list[tuple[str, str | None]]:
    """Find all projects configured to use the same gate path.

    Args:
        gate_path: The gate path to check for
        exclude_project: Project ID to exclude from results (usually the current project)

    Returns:
        List of (project_id, upstream_url) tuples for projects sharing this gate
    """
    from ..core.projects import list_projects as _list_projects

    gate_path = gate_path.resolve()
    return [
        (project.id, project.upstream_url)
        for project in _list_projects()
        if project.id != exclude_project and project.gate_path.resolve() == gate_path
    ]

get_project(project_id)

Load a project by ID and return a rich Project aggregate.

Source code in src/terok/lib/domain/project.py

def get_project(project_id: str) -> Project:
    """Load a project by ID and return a rich [`Project`][terok.lib.domain.project.Project] aggregate."""
    return Project(load_project(project_id))

list_projects()

Return all known projects as rich Project aggregates.

Source code in src/terok/lib/domain/project.py

def list_projects() -> list[Project]:
    """Return all known projects as rich [`Project`][terok.lib.domain.project.Project] aggregates."""
    return [Project(cfg) for cfg in _list_projects()]

project_image_exists(project_id)

Return True when the project's L2 CLI image is present locally.

Source code in src/terok/lib/domain/project.py

def project_image_exists(project_id: str) -> bool:
    """Return ``True`` when the project's L2 CLI image is present locally."""
    return image_exists(project_cli_image(project_id))

summarize_ssh_init(result)

Render an ssh-init result for the terminal.

Source code in src/terok/lib/domain/ssh.py

def summarize_ssh_init(result: SSHInitResult) -> None:
    """Render an ``ssh-init`` result for the terminal."""
    print(f"  id:          {result['key_id']}")
    print(f"  type:        {result['key_type']}")
    print(f"  fingerprint: {result['fingerprint']}")
    print(f"  comment:     {result['comment']}")
    print("Public key (register as a deploy key on the remote):")
    print(f"  {result['public_line']}")

render_project_yaml(values)

Render project.yml without writing it — used by the TUI review screen.

Source code in src/terok/lib/domain/wizards/new_project.py

def render_project_yaml(values: dict) -> str:
    """Render ``project.yml`` without writing it — used by the TUI review screen."""
    variables = {
        "PROJECT_ID": values["project_id"],
        "UPSTREAM_URL": values["upstream_url"],
        "DEFAULT_BRANCH": values["default_branch"],
        "USER_SNIPPET": values["user_snippet"],
        "SECURITY_CLASS": values["security_class"],
        "BASE": values["base"],
        "BASE_IMAGE": BASE_IMAGES[values["base"]],
        # Empty string suppresses the ``agents:`` line via the
        # template's ``{% if AGENTS %}`` gate — the project then
        # inherits the global default written by ``terok agents set``.
        "AGENTS": values.get("agents", ""),
    }
    with resources.as_file(_TEMPLATE_DIR / _TEMPLATE_NAME) as template_path:
        # ``StrictUndefined`` upgrades silent ``{{TYPO}}`` to a hard
        # error; ``autoescape=False`` because YAML output would be
        # corrupted by HTML escaping.  The wizard's template uses
        # ``{% if %}`` blocks and the ``| indent`` filter — Jinja2
        # control flow, not just ``{{VAR}}`` substitution.
        env = jinja2.Environment(  # nosec B701 — see comment above  # noqa: S701
            loader=jinja2.FileSystemLoader(str(template_path.parent)),
            keep_trailing_newline=True,
            undefined=jinja2.StrictUndefined,
            autoescape=False,
        )
        return env.get_template(template_path.name).render(**variables)

validate_answer(question, raw)

Normalise and validate a raw answer for question.

Returns (value, error_or_None) — the normalised value and an error message if the answer was rejected. Both presenters call this so validation semantics stay identical regardless of UI.

Normalisation, in order:

1. Strip surrounding whitespace (copy-paste leftovers, accidental trailing spaces). All-whitespace input is indistinguishable from empty for the required/optional check.
2. Apply question.transform if set (e.g. str.lower).
3. Enforce the required flag against the final value.
4. For kind="choice", the value must be one of the declared slugs — defensive against presenter bugs that might submit a label, index, or free-form typo.
5. Run question.validate for field-specific rules.

Source code in src/terok/lib/domain/wizards/new_project.py

def validate_answer(question: Question, raw: str) -> tuple[str, str | None]:
    """Normalise and validate a raw answer for *question*.

    Returns ``(value, error_or_None)`` — the normalised value and an
    error message if the answer was rejected.  Both presenters call
    this so validation semantics stay identical regardless of UI.

    Normalisation, in order:

    1. Strip surrounding whitespace (copy-paste leftovers, accidental
       trailing spaces).  All-whitespace input is indistinguishable
       from empty for the required/optional check.
    2. Apply ``question.transform`` if set (e.g. ``str.lower``).
    3. Enforce the required flag against the final value.
    4. For ``kind="choice"``, the value must be one of the declared
       slugs — defensive against presenter bugs that might submit a
       label, index, or free-form typo.
    5. Run ``question.validate`` for field-specific rules.
    """
    value = raw.strip()
    if question.transform:
        value = question.transform(value)
    if question.required and not value:
        return value, f"{question.prompt} is required."
    if question.kind == "choice" and value:
        valid_slugs = {slug for slug, _label in question.resolve_choices()}
        if value not in valid_slugs:
            allowed = ", ".join(sorted(valid_slugs))
            return value, f"{question.prompt} must be one of: {allowed}"
    if question.validate:
        err = question.validate(value)
        if err:
            return value, err
    return value, None

write_project_yaml(project_id, rendered, *, overwrite=False)

Write rendered YAML to <user_projects_dir>/<project_id>/project.yml.

The TUI reviews YAML in a TextArea before writing, so this is the write half of generate_config — kept separate so the TUI can pass tweaked content without re-rendering the template.

Source code in src/terok/lib/domain/wizards/new_project.py

def write_project_yaml(project_id: str, rendered: str, *, overwrite: bool = False) -> Path:
    """Write *rendered* YAML to ``<user_projects_dir>/<project_id>/project.yml``.

    The TUI reviews YAML in a ``TextArea`` before writing, so this is the
    write half of [`generate_config`][terok.lib.domain.wizards.new_project.generate_config] — kept separate so the TUI can
    pass tweaked content without re-rendering the template.
    """
    project_dir = user_projects_dir() / project_id
    ensure_dir_writable(project_dir, "Project")
    config_path = project_dir / "project.yml"
    if config_path.exists() and not overwrite:
        return config_path
    config_path.write_text(rendered, encoding="utf-8")
    return config_path