project

project

Rich Project domain object — DDD Aggregate Root.

The central domain object in terok's architecture. Project wraps a ProjectConfig value object with lifecycle behavior and serves as the single entry point for all project-scoped operations:

• Task management — create, list, run, and stop tasks (project.create_task(), project.list_tasks(status="running"))
• Security setup — SSH keypairs (project.ssh) and git gate mirrors (project.gate)
• Agent configuration — layered config resolution and provider selection (project.agents)
• Infrastructure — Dockerfile generation, image builds, and state queries

The object graph follows DDD conventions::

get_project("myproj")         →  Project (Aggregate Root)
    .config                   →  ProjectConfig (Value Object)
    .gate                     →  GitGate (Repository + Gateway)
    .ssh                      →  SSHManager (Service)
    .agents                   →  AgentManager (Strategy + Config Stack)
    .create_task()            →  Task (Entity)
    .get_task(id)             →  Task (Entity)

Subsystems (gate, ssh, agents) are lazy-initialized on first access — constructing a Project performs no I/O beyond loading the config that was already resolved by the caller.

This module also contains delete_project and its helpers, which handle the full teardown of a project including archiving, task cleanup, and safe removal of managed directories.

See Also

get_project — factory that returns a rich Project terok.lib.domain.task — the Task entity contained by Project terok.lib.core.project_model — the ProjectConfig value object

DeleteProjectResult

Bases: TypedDict

Result of a project deletion.

deleted instance-attribute

skipped instance-attribute

archive instance-attribute

ACPEndpoint(project_name, task_id, socket_path, status, bound_agent=None) dataclass

One per-task ACP endpoint as visible from the host.

Constructed by acp_endpoints; consumed by the CLI (terok acp list) and the TUI panel. Carries enough state to render a status row without forcing the listing path to actually probe or open the socket.

project_name instance-attribute

The owning project's name.

task_id instance-attribute

The task this endpoint serves.

socket_path instance-attribute

Where the proxy daemon would bind (or has bound) the socket.

The path is computed deterministically from the task id and may not yet exist on disk — status records whether it does.

status instance-attribute

Live state — active, ready, or unsupported.

bound_agent = None class-attribute instance-attribute

Set only when status == ACTIVE and the daemon has bound an agent for the open session; None otherwise.

AgentManager(config)

Project-scoped agent configuration manager (Strategy + Config Stack).

Resolves the layered agent configuration stack (global → project → CLI overrides) and selects the active headless provider for a project. Used by Project via project.agents.

The config stack is resolved lazily on each call — the manager holds no cached state, so config file changes take effect immediately.

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

__slots__ = ('_config',) class-attribute instance-attribute

resolve_config(cli_overrides=None)

Return the merged agent config dict.

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

def resolve_config(
    self,
    cli_overrides: dict[str, Any] | None = None,
) -> dict[str, Any]:
    """Return the merged agent config dict."""
    return resolve_agent_config(
        self._config.name,
        agent_config=self._config.agent_config,
        project_root=self._config.root,
        cli_overrides=cli_overrides,
    )

resolve_instructions(provider_name)

Return resolved instructions text for the given provider.

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

def resolve_instructions(self, provider_name: str) -> str:
    """Return resolved instructions text for the given provider."""
    effective = self.resolve_config()
    return resolve_instructions(
        effective,
        provider_name,
        project_root=self._config.root,
        family=self._config.known_family,
    )

get_agent(name=None)

Resolve the active headless provider for this project.

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

def get_agent(self, name: str | None = None) -> Agent:
    """Resolve the active headless provider for this project."""
    return get_agent(name, default_agent=self._config.default_agent)

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.name — two Project instances with the same name 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

name property

Return the project name (slug).

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 name.

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

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

__hash__()

Hash by project name for use in sets and dicts.

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

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

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.name, name=name)
    meta = get_task_meta(self._config.name, 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.name, 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.name)
    if mode:
        metas = [m for m in metas if m.mode == mode]
    # Hydrate live container state so status filtering is accurate.
    # A failed query is an error — statuses computed without live state
    # would be wrong, not merely incomplete (#1134).
    live_states = get_all_task_states(self._config.name, metas)
    if live_states is None:
        raise SystemExit("Container runtime unavailable — cannot query container states.")
    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.  Scope-aware: a project with
    # ``credentials.scope: project`` reads its own vault row, not
    # the host-wide bucket — otherwise its tasks would be reported
    # READY on host-wide creds the container will never see.
    # ``stored_credential_entries`` translates the vault's provider
    # keys (anthropic, github) into the agent names the image label
    # speaks — raw vault keys never intersect the label.
    authed = set(stored_credential_entries(self._config.credential_set))
    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.name, task.id)
        sock_exists = sock.exists()
        bound = _read_bound_agent(self._config.name, task.id) if sock_exists else None
        if sock_exists:
            status = ACPEndpointStatus.ACTIVE
        elif _task_has_any_authed_agent(
            self._config.name, task, authed, sandbox=sandbox, label_cache=label_cache
        ):
            status = ACPEndpointStatus.READY
        else:
            status = ACPEndpointStatus.UNSUPPORTED
        out.append(
            ACPEndpoint(
                project_name=self._config.name,
                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.name, 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.name, 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.name)

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.name)

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.name,
        include_dev=include_dev,
        refresh_agents=refresh_agents,
        full_rebuild=full,
    )

state(*, gate_commit_provider=None, gate_pending_provider=None)

Return the project's infrastructure state snapshot.

gate_commit_provider is an optional callable that, given a project name, 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. gate_pending_provider does the same for the count of pending destructive gate ops (the TUI's pending! badge).

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

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

    *gate_commit_provider* is an optional callable that, given a
    project name, 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.
    *gate_pending_provider* does the same for the count of pending
    destructive gate ops (the TUI's ``pending!`` badge).
    """
    from .project_state import get_project_state

    return get_project_state(
        self._config.name,
        gate_commit_provider=gate_commit_provider,
        gate_pending_provider=gate_pending_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.name)

suggested_ssh_key_comment(*, force=False, prompt_on_tty=False)

Suggest a new key's comment, or None when init would reuse a key.

CLI callers may enable the vault's TTY unlock prompt; TUI callers keep it disabled so a locked vault cannot block on terminal input.

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

def suggested_ssh_key_comment(
    self, *, force: bool = False, prompt_on_tty: bool = False
) -> str | None:
    """Suggest a new key's comment, or ``None`` when init would reuse a key.

    CLI callers may enable the vault's TTY unlock prompt; TUI callers
    keep it disabled so a locked vault cannot block on terminal input.
    """
    with vault_db(prompt_on_tty=prompt_on_tty) as db:
        if not force and db.list_ssh_keys_for_scope(self._config.name):
            return None
        return SSHManager(scope=self._config.name, db=db).suggested_comment()

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.name, 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... ")

__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(name={self.name!r}, security={self.security_class!r})"

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 name to exclude from results (usually the current project)	None

Returns:

Type	Description
list[tuple[str, str | None]]	List of (project_name, 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 name to exclude from results (usually the current project)

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

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

validate_gate_upstream_match(project_name)

Validate that no other project uses the same gate with a different upstream.

Raises SystemExit if another project uses the same gate path but has a different upstream_url configured.

Parameters:

Name	Type	Description	Default
project_name	str	The project to validate	required

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

def validate_gate_upstream_match(project_name: str) -> None:
    """Validate that no other project uses the same gate with a different upstream.

    Raises SystemExit if another project uses the same gate path but has a
    different upstream_url configured.

    Args:
        project_name: The project to validate
    """
    project = load_project(project_name)
    sharing = find_projects_sharing_gate(project.gate_path, exclude_project=project_name)

    for other_id, other_url in sharing:
        if other_url is None or project.upstream_url is None or other_url != project.upstream_url:
            this_display = (
                project.upstream_url if project.upstream_url is not None else "<not configured>"
            )
            other_display = other_url if other_url is not None else "<not configured>"
            missing_note = ""
            if other_url is None or project.upstream_url is None:
                missing_note = (
                    "\nNote: One or more projects sharing this gate do not have an "
                    "upstream_url configured in project.yml.\n"
                )
            raise SystemExit(
                f"Gate path conflict detected!\n"
                f"\n"
                f"  Gate path: {project.gate_path}\n"
                f"\n"
                f"  This project ({project_name}):\n"
                f"    upstream_url: {this_display}\n"
                f"\n"
                f"  Conflicting project ({other_id}):\n"
                f"    upstream_url: {other_display}\n"
                f"\n"
                f"Projects sharing a gate must have the same upstream_url.\n"
                f"Either change the gate.path in one project's project.yml,\n"
                f"or ensure both projects point to the same upstream repository.\n"
                f"{missing_note}"
            )

make_git_gate(config, *, use_personal_ssh=None)

Construct a GitGate from a ProjectConfig (adapter factory).

Injects validate_gate_upstream_match as the gate validation callback. The use_personal_ssh flag resolves per-invocation override (e.g. terok gate-sync --use-personal-ssh) > per-project YAML (ssh.use_personal) > default False.

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

def make_git_gate(config: ProjectConfig, *, use_personal_ssh: bool | None = None) -> GitGate:
    """Construct a `GitGate` from a [`ProjectConfig`][terok.cli.commands.sickbay.ProjectConfig] (adapter factory).

    Injects ``validate_gate_upstream_match`` as the gate validation callback.
    The ``use_personal_ssh`` flag resolves per-invocation override (e.g.
    ``terok gate-sync --use-personal-ssh``) > per-project YAML
    (``ssh.use_personal``) > default ``False``.
    """
    effective = use_personal_ssh if use_personal_ssh is not None else config.ssh_use_personal
    return GitGate(
        scope=config.name,
        gate_path=config.gate_path,
        upstream_url=config.upstream_url,
        default_branch=config.default_branch,
        use_personal_ssh=effective,
        validate_gate_fn=validate_gate_upstream_match,
        clone_cache_base=make_sandbox_config().clone_cache_base_path,
        backups_enabled=config.gate_backups_enabled,
        backup_retention_days=config.gate_backup_retention_days,
    )

describe_pending_op(op)

Render one pending destructive gate op as a single decision-ready line.

The operator confirms these sight-unseen otherwise — the line must carry the branch, what would happen, why, and above all whether any gate-local (agent) commits would be discarded.

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

def describe_pending_op(op: PendingOp) -> str:
    """Render one pending destructive gate op as a single decision-ready line.

    The operator confirms these sight-unseen otherwise — the line must
    carry the branch, what would happen, why, and above all whether any
    gate-local (agent) commits would be discarded.
    """
    verb = "delete" if op["kind"] == "delete" else "force-update"
    reasons = {
        "upstream_delete": "deleted upstream",
        "upstream_rewrite": "upstream rewrote history",
        "unknown_provenance": "not on upstream; pre-dates sync tracking",
    }
    if op["lossless"]:
        impact = "no gate-local commits"
    elif op["gate_only_commits"] is not None:
        impact = f"would discard {op['gate_only_commits']} gate-local commit(s)"
    else:
        impact = "gate-local work cannot be ruled out"
    return f"{verb} {op['branch']} ({reasons[op['reason']]}; {impact})"

summarize_gate_sync(result)

Turn a gate sync report into the abridged lines the CLI/TUI print.

Shows what actually happened branch by branch — the whole point of the structured report — while capping each category so a first sync of a thousand-branch repo doesn't scroll the terminal into oblivion.

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

def summarize_gate_sync(result: GateSyncResult) -> list[str]:
    """Turn a gate sync report into the abridged lines the CLI/TUI print.

    Shows what actually happened branch by branch — the whole point of
    the structured report — while capping each category so a first sync
    of a thousand-branch repo doesn't scroll the terminal into oblivion.
    """
    max_lines = 10
    lines: list[str] = []

    def _capped(items: list[str]) -> list[str]:
        if len(items) > max_lines:
            return items[:max_lines] + [f"  … and {len(items) - max_lines} more"]
        return items

    if result["migrated"]:
        lines.append("Gate migrated to the safe sync model (one-time normalisation).")

    by_kind: dict[str, list[AppliedOp]] = {}
    for op in result["applied"]:
        by_kind.setdefault(op["kind"], []).append(op)
    labels = {"create": "created", "fast_forward": "fast-forwarded"}
    for kind, label in labels.items():
        if ops := by_kind.get(kind):
            lines.append(f"{label}:")
            lines += _capped([f"  {o['branch']} -> {(o['new_sha'] or '')[:12]}" for o in ops])

    if gate_only := result["gate_only_branches"]:
        lines.append(f"kept (gate-only, not on upstream): {len(gate_only)}")
        lines += _capped([f"  {branch}" for branch in gate_only])

    if pending := result["pending"]:
        lines.append("pending destructive change(s) — nothing applied without confirmation:")
        lines += _capped([f"  {describe_pending_op(op)}" for op in pending])

    lines += [f"note: {note}" for note in result["notes"]]
    if not result["applied"] and not result["pending"]:
        lines.append("gate is up to date with upstream")
    return lines

make_ssh_manager(config)

Return an SSHManager for config that owns its vault DB.

Use it as a context manager (with make_ssh_manager(cfg) as m: ...); the DB connection closes on exit.

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

def make_ssh_manager(config: ProjectConfig) -> SSHManager:
    """Return an `SSHManager` for *config* that owns its vault DB.

    Use it as a context manager (``with make_ssh_manager(cfg) as m: ...``);
    the DB connection closes on exit.
    """
    return SSHManager.open_for_config(
        scope=config.name, cfg=make_sandbox_config(), prompt_on_tty=True
    )

get_project(project_name)

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

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

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

project_image_exists(project_name)

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_name: str) -> bool:
    """Return ``True`` when the project's L2 CLI image is present locally."""
    return image_exists(project_cli_image(project_name))

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()]

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))

delete_project(project_name)

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_name: 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_name)
    if archive_path is None:
        raise SystemExit(
            f"Project archiving failed for '{project_name}'; aborting deletion to prevent data loss."
        )

    project = load_project(project_name)
    pid = project.name
    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)