> ## Documentation Index
> Fetch the complete documentation index at: https://piloop.autoresearch.inc/llms.txt
> Use this file to discover all available pages before exploring further.

# Executors, stores and services

> Plugins that decide where jobs run, how machines reach the server, and where code, blobs, secrets and memory live.

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

```toml pyproject.toml theme={null}
[project.entry-points."piloop.executors"]
inline = "piloop_inline:Inline"
```

```python piloop_inline/__init__.py theme={null}
"""``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)
        if self.runtime is not None:  # before the row says running: a prompt can come as soon as it does
            self.steer = partial(self.runtime.prompt, handle)
        try:
            self._executor.host.results.started(self.key, handle.data, session_id)
        except Settled:  # its row closed meanwhile: nothing waits for the agent
            self.steer = None
            return False
        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`:

```toml theme={null}
[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](https://piloop.autoresearch.inc/guides/workers.md#capacity) reach it through `configure`, and a [network](#networks) says who may call
the routes of its [contribution](https://piloop.autoresearch.inc/plugins/contributions.md#routes-and-consumers).

| 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](#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](#leases), taken by its worker in `piloop serve` or by
a [worker on another machine](https://piloop.autoresearch.inc/guides/workers.md). Its routes are the lease calls. Its
keys are in the [configuration reference](https://piloop.autoresearch.inc/reference/configuration.md#plugins).

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

```mermaid theme={null}
stateDiagram-v2
    direction LR
    [*] --> queued: the executor submits the job
    queued --> acquired: a worker takes it
    acquired --> running: the job starts
    running --> running: heartbeat
    running --> reported: the worker reports
    acquired --> expired: silent past the toleration
    running --> expired: silent past the toleration
    queued --> cancelled: the daemon stops first
```

| 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](https://piloop.autoresearch.inc/guides/workers.md) 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:

```python theme={null}
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)
```

```toml theme={null}
[project.entry-points."piloop.networks"]
trusted = "my_package:Trusted"
```

```text theme={null}
[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](https://piloop.autoresearch.inc/reference/configuration.md#plugins).

## 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](https://piloop.autoresearch.inc/guides/workers.md#workers-that-the-server-starts) sets one up.

### Example: workers in Docker containers

```python theme={null}
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:

```toml pyproject.toml theme={null}
[project.entry-points."piloop.cas"]
sqlite = "piloop_sqlite_cas:Sqlite"
```

```python piloop_sqlite_cas/__init__.py theme={null}
"""``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`):

```toml theme={null}
[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:

```python theme={null}
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
```

```toml theme={null}
[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`. The caller closes it. |
| `Vcs.clone(url: str, path: Path, credential: SecretStr \| None) -> Repository` | Clone the repository at `url` into `path`, with every version that piloop keeps there, 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.agent_env(agent: Agent) -> dict[str, str]` | The variables of `agent`'s job that the provider sets: who the versions that it saves are by. Empty: none. |
| `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.close() -> None` | Release what the open repository holds. Later calls do nothing. |
| `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]) -> Captured` | Put the versions that an agent saved in this checkout of `base` into `cas`. |

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

```python theme={null}
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
```

```toml theme={null}
[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. |

## Notifiers

A **notifier** sends the inbox items that need a human to a service outside piloop. `[plugins.notifier.<name>]` in
`config.toml` configures one, and `[plugins.notifier] use` names those in use. `slack`, built in, posts to an incoming
webhook. [Notifications](https://piloop.autoresearch.inc/guides/notifications.md) shows its keys. A notifier sends every notice in one message, or raises
`NotDelivered`: piloop records which items it sent, retries, and reminds.

```python theme={null}
from collections.abc import Sequence
from urllib.error import URLError
from urllib.request import urlopen

from piloop.capabilities import HomeSecrets, NotDelivered, Notice, Notifier


class Ntfy(Notifier):
    """``ntfy``: one push to an ntfy.sh topic."""

    topic: str

    def send(self, notices: Sequence[Notice], secrets: HomeSecrets) -> None:
        text = '\n'.join(f'{n.campaign}: {n.comment_kind} comment#{n.comment}' for n in notices)
        try:
            urlopen(f'https://ntfy.sh/{self.topic}', data=text.encode(), timeout=10).close()
        except URLError as e:
            raise NotDelivered(str(e)) from e
```

```toml theme={null}
[project.entry-points."piloop.notifiers"]
ntfy = "my_package:Ntfy"
```

| Member | What it is |
| - | - |
| `Notifier.link: str \| None` | The page of an item, such as `https://piloop.example.com/inbox/{comment}`. `{comment}` is the item's id and `{campaign}` its campaign's name. Unset: a message names the command that shows the item. |
| `Notifier.remind_every: timedelta \| Literal['never']` | How long after a send an item that nobody acknowledged is sent again. `"never"`: once. |
| `Notifier.retry: timedelta` | How long after a failed send the next try comes. It doubles with each failure in a row, up to `retry_max`. |
| `Notifier.retry_max: timedelta` | The longest wait between two tries. |
| `Notifier.send(notices: Sequence[Notice], secrets: HomeSecrets) -> None` | Send `notices` in one message. `secrets` has the credentials that the notifier's keys name. |
| `Notifier.reminder(sent: datetime) -> datetime \| None` | When an item that was `sent` then, and that still waits, is sent again: None for `never`. |
| `Notifier.wait(failures: int, asked: timedelta \| None) -> timedelta` | How long to wait after the `failures`-th failure in a row: what the receiver `asked`, else the backoff. |
| `Notice.kind: NoticeKind` | Whether the notifier sends it for the first time, or again. |
| `Notice.comment: CommentId` | The item: `piloop inbox show <comment>` shows it. |
| `Notice.comment_kind: CommentKind` | What the item is: an `escalation`, a `decision`, … |
| `Notice.priority: int` | 0 (urgent) to 4 (someday). |
| `Notice.campaign: str \| None` | The campaign's name, or None for an item of no campaign. |
| `Notice.author: str` | Who wrote the item: its signature, such as `implementer#12`. |
| `Notice.body: str` | The item's text, in full. |
| `Notice.owner: Person \| None` | Whoever added the campaign, when a person did: whom the item concerns on a shared server. |
| `Notice.link: str \| None` | The item's page, from the notifier's `link`. None when it has none: show `piloop inbox show <comment>`. |

## Archive

The **archive** moves the finished parts of campaign workspaces into the CAS, and writes them back on
`piloop restore`. `[plugins.archive]` in `config.toml` selects it. `local`, built in and the default, records each
archiving in its `archives` table. Its archiver is a [consumer](https://piloop.autoresearch.inc/plugins/contributions.md#routes-and-consumers) of the
daemon's loop. [The archive and the CAS](https://piloop.autoresearch.inc/concepts/archive.md) describes what it does. `local`, in
[`piloop/plugins/archive/local/`](https://github.com/hiloopai/piloop/tree/main/piloop/plugins/archive/local), is
the model for another provider.

| 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`, and its `Lifecycle`: the daemon's stop waits for the job in flight. |

## 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](https://piloop.autoresearch.inc/plugins/contributions.md) shows how. Its contract has one method,
`sync`. [Memory](https://piloop.autoresearch.inc/concepts/memory.md) describes it.

This provider is `local` with other default documents:

```python theme={null}
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']
```

```toml theme={null}
[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.


This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.