Executors, stores and services

These capabilities are host capabilities: their providers get live objects of piloop. Most are the home’s choice, in $PILOOP_HOME/config.toml, and take effect at piloop restart. The VCS and memory are a campaign’s, in piloop.toml.

Executors

An executor decides where each job runs: an agent session or an evaluation. The daemon submits the job, and the executor reports how it ended.

Example: run each job in a thread of the daemon

pyproject.toml

[project.entry-points."piloop.executors"]
inline = "piloop_inline:Inline"

piloop_inline/__init__.py

"""``inline``: a piloop executor that runs each job at once, in a thread of the daemon, with no leases.

Each job runs as the built-in ``workers`` executor's worker runs it (``run_agent``, ``run_evaluation``), and its result
goes to core through ``Host.results``. There is no fence: one thread owns each job, and only
it reports. Nothing survives the daemon: ``adopt`` takes over no job, and ``hand_over`` stops every job.
"""

import subprocess
import threading
from collections.abc import Callable, Sequence
from contextlib import suppress
from functools import partial

from covalence import Cid

from piloop.capabilities import (
    AgentJob,
    AgentKey,
    AgentReport,
    Capacity,
    Contribution,
    EvaluationJob,
    EvaluationReport,
    EventModel,
    ExecutorConfig,
    Handle,
    HarnessSession,
    Host,
    JobKey,
    JobSecrets,
    LeftJob,
    Prompt,
    Runtime,
    Settled,
    job_secrets,
    key_of,
    kill_group,
    local_site,
    remove,
    run_agent,
    run_evaluation,
)


class Inline(ExecutorConfig):
    """``inline``: it takes no parameters."""

    def open(self, host: Host) -> InlineExecutor:
        return InlineExecutor(host)


class InlineExecutor:
    """``Executor``: a thread for each job, started when the job is submitted."""

    def __init__(self, host: Host) -> None:
        self.host = host
        self.site = local_site(host.home, host.cas, host.repos, host.records, host.reserve)
        self._lock = threading.Lock()
        self._jobs: dict[JobKey, _Job] = {}
        self.stopping = False

    def configure(self, config: ExecutorConfig) -> None:
        """Nothing to change: it takes no parameters."""

    def capacity(self) -> Capacity:
        """No bound: every job runs as soon as it is submitted."""
        return Capacity(agents=None, evaluations=None)

    def submit(self, job: AgentJob | EvaluationJob) -> None:
        running = _Job(self, job)
        with self._lock:
            if self.stopping:
                raise RuntimeError('the executor stops: it takes no more jobs')
            self._jobs[running.key] = running
        running.thread = threading.Thread(target=self._run, args=(job, running), daemon=True)
        running.thread.start()

    def prompt(self, key: AgentKey, prompt: Prompt) -> None:
        """Type the prompt into the agent's runtime at once, and say how that went."""
        with self._lock:
            running = self._jobs.get(key)
        steer = running.steer if running is not None else None
        error = None if steer is not None else 'the agent is not running'
        if steer is not None and running is not None:
            try:
                steer(prompt.text)
            except Exception as e:  # the runtime refused it (a headless agent): core records why
                error = str(e)
            else:
                running.prompts.append([prompt])
        self.host.results.queued(key, prompt.id, error)

    def cancel(self, key: JobKey) -> None:
        with self._lock:
            running = self._jobs.get(key)
        if running is not None:
            running.stop()

    def adopt(self, left: Sequence[LeftJob]) -> list[JobKey]:
        """Take over nothing: a previous daemon's jobs died with its threads."""
        return [job.key for job in left]

    def contribution(self) -> Contribution:
        """Nothing: a thread reports its own job, so it has no upkeep, and every job runs in the daemon, so no route."""
        return Contribution()

    def capacity_events(self) -> frozenset[type[EventModel]]:
        """None: a slot frees only when a job ends."""
        return frozenset()

    def stop(self, timeout: float) -> None:
        with self._lock:
            self.stopping = True
            running = list(self._jobs.values())
        for job in running:
            job.stop()
        for job in running:
            if job.thread is not None:
                job.thread.join(timeout)

    def hand_over(self) -> None:
        """Stop every job: a next daemon could not find them again."""
        self.stop(0)

    def close(self) -> None:
        """Nothing to release: the threads end with their jobs."""

    def _run(self, job: AgentJob | EvaluationJob, running: _Job) -> None:
        """Run ``job`` to its end and apply its result, or tell core it is lost when the run itself fails."""
        results, key = self.host.results, running.key
        try:
            with self.host.tracer.span('piloop.job', job=job.kind) as span:
                if isinstance(job, AgentJob):
                    runtime = self.host.runtimes(job.runtime)
                    running.runtime = runtime
                    report: AgentReport | EvaluationReport | None = run_agent(job, runtime, running, self.site, span)
                    running.steer = None  # the agent ended: a prompt from now on is undelivered
                else:
                    report = run_evaluation(job, running, self.site, span)
            if report is not None:
                with suppress(Settled):  # its row closed meanwhile: a stop
                    results.finished(job, report)
            if isinstance(job, AgentJob):
                remove(job, self.site)
        except Exception as e:  # a fault of the executor: nothing must wait for this job forever
            with suppress(Settled):
                results.lost(key, f'{type(e).__name__}: {e}')
        finally:
            with self._lock:
                self._jobs.pop(key, None)


class _Job:
    """``Control`` for one job: how to stop it, and its start and session id."""

    def __init__(self, executor: InlineExecutor, job: AgentJob | EvaluationJob) -> None:
        self._executor = executor
        self.job = job
        self.key = key_of(job)
        self.thread: threading.Thread | None = None
        self.runtime: Runtime | None = None
        self.steer: Callable[[str], None] | None = None
        """How to prompt the agent, once it started."""
        self.prompts: list[list[Prompt]] = []
        """What its runtime took, one prompt at a time: the report says which ones the agent read."""
        self._kill: Callable[[], None] | None = None
        self._asked = False

    def stop(self) -> None:
        self._asked = True
        if (kill := self._kill) is not None:
            kill()

    def begin(self) -> bool:
        return not self._executor.stopping

    def started(self, handle: Handle, session_id: str | None) -> bool:
        assert isinstance(self.key, AgentKey)
        try:
            self._executor.host.results.started(self.key, handle.data, session_id)
        except Settled:  # its row closed meanwhile: nothing waits for the agent
            return False
        if self.runtime is not None:
            self.steer = partial(self.runtime.prompt, handle)
        return True

    def running(self, kill: Callable[[], None] | None) -> None:
        self._kill = kill
        if kill is not None and self._asked:  # a cancel came while it started
            kill()

    def process(self, proc: subprocess.Popen[bytes]) -> bool:
        self._kill = partial(kill_group, proc)
        return not self._asked and not self._executor.stopping

    def session(self, session_id: str) -> None:
        assert isinstance(self.key, AgentKey)
        with suppress(Settled):
            self._executor.host.results.session(self.key, HarnessSession(id=session_id))

    def saved(self, session_file: Cid) -> None:
        assert isinstance(self.key, AgentKey)
        with suppress(Settled):
            self._executor.host.results.session(self.key, HarnessSession(file=session_file))

    def asked(self) -> bool:
        return self._asked

    def stopping(self) -> bool:
        return self._executor.stopping

    def handing_over(self) -> bool:
        return False

    def queued(self) -> list[list[Prompt]]:
        return self.prompts

    def secrets(self) -> JobSecrets:
        assert isinstance(self.job, AgentJob)
        host = self._executor.host
        with host.db.session() as s:
            found = job_secrets(s, host.secrets, self.job)
            s.commit()
        return found

Select it in $PILOOP_HOME/config.toml, then run piloop restart:

[plugins.executor]
provider = "inline"

The contract

The entry point names a subclass of ExecutorConfig: its fields are the other keys of [plugins.executor]. The daemon calls its open(host) once, when it starts, to make the Executor. The live limits reach it through configure, and a network says who may call the routes of its contribution.

Member	What it is
`ExecutorConfig.open(host: Host) -> Executor[Self]`	Make the executor. The daemon calls it once, when it starts.
`Executor.capacity() -> Capacity`	How many more jobs of each kind it can take now: its free slots, less the jobs it has and has not started.
`Executor.capacity_events() -> frozenset[type[EventModel]]`	The events of this executor that say its capacity grew: core then launches what waits for a slot.
`Executor.configure(config: C) -> None`	Take `config`, the one that opened it with the live limits applied. The daemon calls it when one changes.
`Executor.submit(job: AgentJob \| EvaluationJob) -> None`	Run `job`, now or when a slot is free. Every job ends with `Results.finished` or `Results.lost`.
`Executor.prompt(key: AgentKey, prompt: Prompt) -> None`	Give `prompt` to the running agent’s runtime at least once, then call `Results.queued`.
`Executor.cancel(key: JobKey) -> None`	Stop a job: a kill or a cancel. It still reports how it ended. Called again, it tries again.
`Executor.adopt(left: Sequence[LeftJob]) -> list[JobKey]`	Take over the jobs that a previous daemon process left, and return the keys of those it cannot.
`Executor.stop(timeout: float) -> None`	Stop every job, and wait up to `timeout` for them to report. Submit no more.
`Executor.hand_over() -> None`	Leave the jobs running for the next daemon process (an in-place restart), which adopts them.
`Executor.close() -> None`	Release what the executor holds. The daemon calls it last.

A job’s end goes to host.results: finished or lost, in a session that the executor commits. Core raises Settled for a job whose row is closed already. An executor whose workers can report late uses leases to refuse those reports; inline needs none. run_agent and run_evaluation run one job on this machine.

Built in

workers, the default, runs every job on a lease, taken by its worker in piloop serve or by a worker on another machine. Its routes are the lease calls. Its keys are in the configuration reference.

Leases

A lease is a time-limited claim on one job by one worker. Any executor can use leases: Leases in piloop.capabilities. workers runs every job on one.

Word	Meaning
Job	One agent session or one evaluation: what to check out, what to run, what to bring back. It is a blob in the CAS, and its lease names it by address. Its commit, its inputs and its results go by address too.
Worker	A process that takes leases and runs their jobs. `piloop serve` has one, called `serve`.
Heartbeat	The worker renews its leases every `renew_seconds`.
Fencing token	The lease id. It grows with each new lease of a job, and a report with an old one is refused.
Revoke	A kill or a cancel: the worker stops the job, and the job still reports.
Message	An event that a worker reads after its cursor, with a long-poll: a job waits, stop a job, or a prompt for an agent. A restart of either side loses none.

Workers sets up pools and says what happens when a worker goes silent.

Networks

A network says how other machines reach the server of piloop serve --listen, and who each caller is. Only an executor with routes needs one.

Example: trust a tailnet

This network is https, and it lets every caller from one address range act without a token:

from collections.abc import Mapping
from ipaddress import IPv4Network, IPv6Network, ip_address, ip_network

from piloop.capabilities import Access, Caller, Https, HttpsNetwork, NetworkHost, Peer, Refused


class Trusted(Https):
    """``trusted``: ``https``, and a caller from ``cidr``, such as a tailnet, may act without a token."""

    cidr: str = '100.64.0.0/10'

    def open(self, host: NetworkHost) -> HttpsNetwork:
        return TrustedNetwork(host, self, ip_network(self.cidr))


class TrustedNetwork(HttpsNetwork):
    def __init__(self, host: NetworkHost, config: Https, trusted: IPv4Network | IPv6Network) -> None:
        super().__init__(host, config)
        self.trusted = trusted

    def identify(self, headers: Mapping[str, str], peer: Peer) -> Caller | Refused:
        if peer.address is not None and ip_address(peer.address.host) in self.trusted:
            return Caller(name=peer.address.host, access=Access.ACT)
        return super().identify(headers, peer)

[project.entry-points."piloop.networks"]
trusted = "my_package:Trusted"

[plugins.network]
provider = "trusted"
listen = "0.0.0.0:8443"
tls = "proxy"

The contract

The entry point names a subclass of NetworkConfig: its fields are the other keys of [plugins.network]. Its client finds this machine’s credential itself, or raises NoCredential with what to set: piloop worker dials with it. The daemon calls its open(host) once, when it starts, to make the server’s Network. A request that identify refuses gets 401.

Member	What it is
`NetworkConfig.listen: Address \| None`	Where the server listens, as `HOST:PORT`. None: no server. `piloop serve --listen` overrides it.
`NetworkConfig.CREDENTIALS: ClassVar[frozenset[str]]`	The variables that hold the credential that `client` reads: `piloop worker` withholds them from its jobs.
`NetworkConfig.open(host: NetworkHost) -> Network`	Make the server’s side of the network. The daemon calls it once, when it starts.
`NetworkConfig.client(server: str, actor: Actor \| None = None) -> httpx.Client`	The dialing side, which needs no host: a client of the server at `server`, with this machine’s credential.
`NetworkConfig.listening(listen: Address \| None) -> NetworkConfig`	This network, with `listen` (`piloop serve --listen`) in place of its own when it is not None.
`Network.takes_tokens: bool`	Whether core admits piloop’s own tokens (`plt_`) on this network’s listener. False: the network authenticates every caller itself (`identify`), and core refuses a `plt_` bearer there, so no token minted in the home opens a way around it. The home’s Unix socket verifies them whatever this says.
`Network.listener() -> Listener \| None`	Bind `listen`, or return None when it is None. The daemon calls it once.
`Network.identify(headers: Mapping[str, str], peer: Peer) -> Caller \| Refused`	Who sends a request with these headers (their names in lower case), from `peer`.
`Network.actor(headers: Mapping[str, str]) -> str \| None`	Whom a request asks to act as: an actor’s signature, such as `orchestrator`, or None for the caller itself.
`Network.enroll(principal: str, access: Access) -> Enrollment`	Give the machine called `principal` a credential of `access`, in place of any that it had.
`Network.revoke(principal: str) -> None`	End the credential of `principal`: its next call is refused. One that is gone already is a success.

Built in

https, the default, knows no caller other than the tokens of piloop token, and its client sends the one in PILOOP_HTTPS_TOKEN. An address that is not loopback needs tls: a certificate, or "proxy" when a reverse proxy or an SSH tunnel ends TLS. Its keys are in the configuration reference.

Provisioners

A provisioner starts and stops workers on a platform, for a pool whose jobs wait. A pool’s [plugins.executor.pools.<name>.provisioner.compute] table selects it: there is no default. Workers that the server starts sets one up.

Example: workers in Docker containers

import os
import subprocess
from collections.abc import Sequence

from piloop.capabilities import Instance, InstanceState, Launch, ProvisionerConfig, ProvisionerHost, Provisioning


class Docker(ProvisionerConfig):
    """``docker``: each worker is a container of ``image`` on this machine."""

    image: str

    def open(self, host: ProvisionerHost) -> Provisioning:
        return Containers(self.image, host.owner)


class Containers:
    def __init__(self, image: str, owner: str) -> None:
        self.image, self.owner = image, owner

    def list(self) -> Sequence[Instance]:
        listed = subprocess.run(
            ['docker', 'ps', '-a', '--filter', f'label=piloop.owner={self.owner}',
             '--format', '{{.Names}} {{.Label "piloop.pool"}} {{.State}}'],
            capture_output=True, text=True, check=True,
        ).stdout.split('\n')
        states = {'created': InstanceState.PENDING, 'running': InstanceState.RUNNING}
        return [
            Instance(name=name, pool=pool, state=states.get(state, InstanceState.ENDED), handle={})
            for name, pool, state in (line.split() for line in listed if line)
        ]

    def start(self, launch: Launch) -> None:
        env = {name: value.get_secret_value() for name, value in launch.env.items()}  # never in the argv
        subprocess.run(
            ['docker', 'run', '-d', '--name', launch.name, '--label', f'piloop.owner={self.owner}',
             '--label', f'piloop.pool={launch.pool}', *(f'--env={name}' for name in env), self.image, *launch.argv],
            env=os.environ | env, check=True,
        )

    def stop(self, name: str) -> None:
        subprocess.run(['docker', 'rm', '-f', name], capture_output=True, check=False)

The worker’s credential goes to docker in its environment, never in its argv. Entry point group: piloop.provisioners. The provider’s open makes its live side, which lists, starts and stops instances. It tags each instance with the server’s id, and lists only those, so two servers on one account never stop each other’s.

Member	What it is
`ProvisionerConfig.open(host: ProvisionerHost) -> Provisioning`	Make the live side. The executor calls it once, when the daemon starts.
`Provisioning.list() -> Sequence[Instance]`	Every instance tagged with this server’s owner, in any pool, that the platform still has.
`Provisioning.start(launch: Launch) -> None`	Start the instance `launch.name`. Started already: nothing. Raises `StartRefused`.
`Provisioning.stop(name: str) -> None`	Stop the instance `name`. An instance that is gone already is a success.

modal is built in. It needs the modal extra: uv tool install --reinstall 'piloop[modal]'.

CAS providers

The CAS is the content-addressed store that archived files and manifests live in. A provider decides where its blobs go: a directory, a bucket, or a cache over another provider.

Example: covalence’s SQLite store

This provider keeps the blobs in one SQLite file. Its store is covalence’s Cas.sqlite, as it is:

pyproject.toml

[project.entry-points."piloop.cas"]
sqlite = "piloop_sqlite_cas:Sqlite"

piloop_sqlite_cas/__init__.py

"""``sqlite``: a piloop CAS provider whose store is covalence's ``Cas.sqlite``, all the blobs in one SQLite file."""

from pathlib import Path

from covalence import Cas as SqliteCas

from piloop.capabilities import Cas, CasProvider


class Sqlite(CasProvider):
    """``[plugins.cas] provider = "sqlite"``: the blobs in one SQLite file."""

    path: Path
    """The database file. It is created at the first put."""

    def open(self, home: Path) -> Cas:
        return SqliteCas.sqlite(self.path)

    def where(self, home: Path) -> str:
        return f'sqlite://{self.path}'

Install it, then select it in $PILOOP_HOME/config.toml and restart the daemon (piloop restart):

[plugins.cas]
provider = "sqlite"
path = "/var/lib/piloop/blobs.sqlite"

The contract

A provider is a pydantic model, a subclass of CasProvider: the name of its entry point is its provider, and its fields are the other keys of [plugins.cas]. A key whose value is another provider’s table, as the remote of cached, is a Annotated[CasProvider, CAS.entry] field. The store is a Cas: the methods of covalence’s Python Cas that piloop calls, with the same signatures, so a covalence.Cas is a store as it is. A blob’s address is a raw BLAKE3 CID of its content. A missing blob reads as None, and bytes that do not match their CID raise covalence’s IntegrityError. A store with a local tier over a remote one also implements Upkeep, which piloop store and the daemon’s hourly upkeep use. A store that can be such a local tier is Evictable. Its provider is a LocalProvider, and an Annotated[LocalProvider, CAS.entry] key, as cached’s local, takes only those.

Member	What it is
`CasProvider.open(home: Path) -> Cas`	The store that these parameters describe, for the PILOOP_HOME `home`.
`CasProvider.where(home: Path) -> str`	Where its blobs are, in one spelling (`s3://bucket/prefix`): a cache over it records which store it fills.
`Cas.has(cid: Cid) -> bool`	Whether the store holds the blob, without checking its bytes.
`Cas.missing(cids: Sequence[Cid]) -> list[Cid]`	Those of `cids` that the store lacks, in order.
`Cas.put(data: bytes) -> Cid`	Store `data` and return its raw BLAKE3 CID.
`Cas.get(cid: Cid) -> bytes \| None`	The bytes under `cid`, checked, or None if missing.
`Cas.put_file(path: str \| PathLike[str]) -> Cid`	Store the file at `path`, read twice (to hash, to store), and return its raw BLAKE3 CID.
`Cas.get_file(cid: Cid, path: str \| PathLike[str]) -> int \| None`	Write the blob to `path`, checked in a temporary file beside it that then replaces it; its length.
`Cas.put_fileobj(file: BinaryIO, *, max_bytes: int, expected_length: int \| None = None) -> Cid`	Store a seekable file object, read twice from its start, and return its raw BLAKE3 CID.
`Cas.get_to(cid: Cid, writer: Writer, *, max_bytes: int, expected_length: int \| None = None) -> int \| None`	Write the blob to `writer` and return its length, or None if missing.
`Upkeep.usage() -> TierUsage`	Count the blobs of the local tier: one scan of it.
`Upkeep.flush(*, evict: bool) -> Flushed`	Upload what the remote tier lacks; with `evict`, then drop the local copies that the policy says to.
`Upkeep.integrity(cid: Cid) -> Integrity`	Read and hash a local copy; ask the remote tier about an evicted blob, without fetching it.
`Evictable.held() -> Iterator[Held]`	Every blob whose bytes the tier holds: one scan of it.
`Evictable.release(cid: Cid) -> int`	Drop the tier’s bytes of `cid`; the bytes freed, 0 when it held none.

ObjectsCas (piloop.plugins.cas.kit.objects) is a store of compressed objects over anything that puts, opens and finds whole files by key: disk, s3 and gcs are three. check_cas(cas, scratch) runs the contract against a store, as piloop’s tests do with each built-in provider and with covalence’s Cas.memory(). It never deletes, so point it at a store that you can empty.

Built in

Provider	What it is	Keys
`cached`	A local tier over `remote`, evicted by age and by size.	`remote`, `local`, `evict_after_days`, `local_max_gb`
`disk`	The blobs in a directory.	`path`
`gcs`	A bucket, a prefix in it, and the storage class of each blob.	`bucket`, `prefix`, `storage_class`, `project`, `credentials_file`
`http`	A CAS service, dialed through this home’s network.	`url`
`s3`	A bucket and prefix, and the storage class of each blob.	`url`, `storage_class`, `standard_below`

VCS

The VCS provider keeps a campaign’s versions: it saves, merges, checks out and moves them between machines through the CAS. [plugins.vcs] in piloop.toml selects it. git, built in, is the default and the only one. Most changes start from git and change one method. This one adds a line to the implementers’ brief on how to save their work:

from piloop.capabilities import Git, Part


class SignedGit(Git):
    """``signed_git``: ``git``, and the implementers sign each commit."""

    def prompt(self, part: Part) -> str:
        text = super().prompt(part)
        return text + '\n\nSign each commit: `git commit -S`.' if part is Part.SAVE else text

[project.entry-points."piloop.vcs"]
signed_git = "my_package:SignedGit"

Member	What it is
`Vcs.init(path: Path, summary: str, owned: Sequence[str], ignore: Sequence[str]) -> list[str]`	Make `path` a repository if it is not one, and save its files for `piloop init`.
`Vcs.open(path: Path) -> Repository`	The repository of the checkout at `path`.
`Vcs.clone(url: str, path: Path, credential: SecretStr \| None) -> Repository`	Clone the repository at `url` into `path`, and return it.
`Vcs.replica(path: Path) -> Repository`	A worker’s replica of a repo of record at `path`, made empty if none is there: `Repository.fetch` fills it.
`Vcs.identity(path: Path) -> str`	The project that the checkout at `path` is, the same for each clone of it: `campaigns.repo`.
`Vcs.root(path: Path) -> Path \| None`	The main checkout that `path` is in, also from a checkout that `Repository.checkout` made. None: none.
`Vcs.changed(path: Path, ignore: Sequence[str]) -> bool`	Whether an agent changed the checkout at `path`: a saved version, or a changed file outside `ignore`.
`Vcs.prompt(part: Part) -> str`	What the built-in prompts say about `part` for this provider (`prompted`).
`Vcs.checks() -> list[Check]`	Check that the provider can work on this machine. `piloop doctor` shows the result. Read only.
`Repository.head() -> Version`	The version that the checkout is at.
`Repository.version(native: str) -> Version`	The version that `native` names in the provider’s own terms: a revision, or under `git` a branch or tag.
`Repository.receive(cas: Cas, transfer: Transfer, base: Revision) -> list[Version]`	Take the versions of `transfer` from `cas`, and return those from `base` to its head, oldest first.
`Repository.send(cas: Cas, revisions: Sequence[Revision], have: Sequence[Revision]) -> Cid \| None`	Put what `revisions` have beyond `have` into `cas`, as a `Transfer`’s blob, or None when nothing is.
`Repository.fetch(cas: Cas, revisions: Sequence[Revision], ask: Fetch) -> None`	Take those of `revisions` that the repository lacks: `ask` puts them into `cas`.
`Repository.merge(onto: Revision, other: Revision, message: str) -> Version \| Conflicts`	Save the merge of `other` onto `onto` as a version, without a checkout, or return what conflicts.
`Repository.contains(revision: Revision, ancestor: Revision) -> bool`	Whether `revision` has all of `ancestor`: it descends from it, or merging it in changes nothing.
`Repository.keep(revision: Revision, why: Keep) -> None`	Keep `revision` for `why`, also after whatever named it is gone.
`Repository.kept(campaign: str, run: RunId) -> list[Version]`	The versions of `run` that `keep` kept (`RunVersion`), in their order, for a run before 0.3.0.
`Repository.checkout(revision: Revision, path: Path) -> None`	Put the files of `revision` at `path`, where an agent or an evaluator works on them.
`Repository.discard(path: Path) -> None`	Remove what `checkout` put at `path`. Nothing fails if it is gone, or a directory of something else.
`Repository.checkouts(root: Path) -> list[Path]`	The paths under `root` that `checkout` made and `discard` did not remove yet.
`Repository.untracked(revision: Revision, names: Sequence[str]) -> list[str]`	Those of `names` that `revision` does not hold: what was put in its checkout from elsewhere.
`Repository.capture(cas: Cas, base: Revision, run: RunId, ignore: Sequence[str]) -> Transfer \| None`	Put the versions that an agent saved in this checkout of `base` into `cas`, or None when it saved none.

Secrets

The secrets provider stores the values of piloop secret set. [plugins.secrets] in config.toml selects it. local, built in and the default, keeps them in the home’s database. piloop records who set a secret and when as events, so a provider stores only the values.

import os
from collections.abc import Collection

from pydantic import SecretStr
from sqlalchemy.orm import Session

from piloop.capabilities import LocalSecrets, Owner


class EnvFallback(LocalSecrets):
    """``env_fallback``: ``local``, and a home secret that it lacks comes from the daemon's environment."""

    def get(self, s: Session, owner: Owner, names: Collection[str]) -> dict[str, SecretStr]:
        found = super().get(s, owner, names)
        if owner is None:
            found |= {name: SecretStr(os.environ[name]) for name in names if name not in found and name in os.environ}
        return found

[project.entry-points."piloop.secrets"]
env_fallback = "my_package:EnvFallback"

Member	What it is
`SecretsProvider.put(s: Session, owner: Owner, name: EnvName, value: SecretStr) -> None`	Store `value` as `owner`’s secret `name`, in place of any. The caller commits.
`SecretsProvider.delete(s: Session, owner: Owner, name: EnvName) -> bool`	Delete `owner`’s secret `name`, and return whether there was one. The caller commits.
`SecretsProvider.names(s: Session) -> list[SecretKey]`	Every secret, by owner (the home’s first) and name.
`SecretsProvider.get(s: Session, owner: Owner, names: Collection[EnvName]) -> dict[EnvName, SecretStr]`	The values of those of `names` that `owner` has. Only `values` calls it.

Member	What it is
`ArchiveConfig.open(host: ArchiveHost) -> Archiver`	Make the archiver. The daemon calls it once, when it starts.
`ArchiveConfig.restore(s: Session, cas: Cas, home: Path, campaign: Campaign, target: TargetRef \| str \| None) -> Restored`	Write archived files back into `campaign`’s workspace: a subject’s, a workspace-relative path’s, or all.
`ArchiveConfig.summary(s: Session) -> Summary`	What the archive holds and what waits for its archiving, for `piloop store status`.
`Archiver.contribution() -> Contribution`	Its consumer `CONSUMER`.
`Archiver.stop(timeout: float) -> None`	Stop, and wait up to `timeout` seconds for the job in flight. The daemon’s stop calls it.

Memory

The memory provider keeps what past campaigns leave for the next ones. [plugins.memory] in piloop.toml selects it. local, built in and the default, has the lessons, documents and journal tables, the piloop journal command and its route: Tables, commands, routes and consumers shows how. Its contract has one method, sync. Memory describes it. This provider is local with other default documents:

from piloop.capabilities import LocalMemory

class Evidence(LocalMemory):
    """``evidence``: ``local``, and the documents are the reports under ``docs/evidence/`` unless the table says else."""

    documents: list[str] = ['docs/evidence/**/*.md']

[project.entry-points."piloop.memory"]
evidence = "my_package:Evidence"

Member	What it is
`MemoryProvider.sync(s: Session, campaign: Campaign, actor: Actor) -> Synced`	Take in the campaign’s documents. `piloop init` (campaign registration) and a campaign’s stop call it.

Campaign types

A campaign type is what one kind of campaign stores, and campaigns.type names it. optimization, built in, is the research model: its schema has the ideas, runs, candidates and evaluations. Every campaign is of this type. Its contract has no methods.

Get started

Concepts

Guides

Plugins

Reference

Executors, stores and services

Executors

Example: run each job in a thread of the daemon

The contract

Built in

Leases

Networks

Example: trust a tailnet

The contract

Built in

Provisioners

Example: workers in Docker containers

CAS providers

Example: covalence’s SQLite store

The contract

Built in

VCS

Secrets

Archive

Memory

Campaign types

​Executors

​Example: run each job in a thread of the daemon

​The contract

​Built in

​Leases

​Networks

​Example: trust a tailnet

​The contract

​Built in

​Provisioners

​Example: workers in Docker containers

​CAS providers

​Example: covalence’s SQLite store

​The contract

​Built in

​VCS

​Secrets

​Archive

​Memory

​Campaign types

Executors

Example: run each job in a thread of the daemon

The contract

Built in

Leases

Networks

Example: trust a tailnet

The contract

Built in

Provisioners

Example: workers in Docker containers

CAS providers

Example: covalence’s SQLite store

The contract

Built in

VCS

Secrets

Archive

Memory

Campaign types